// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: authentication-free shared key and token derivation between clients
// feature: add multi-domain support for discovery server
// feature: DServer support simple entity
// feature: suport super-client
// feature: support auth token
// feature: provide query interfaces
// feature: develop dispatcher
// feature: content filter topic
// feature: Split history as an independent common module
// feature: Unified Transport Abstraction Layer Unified Transport Abstraction Layer
// feature: discovery support client and server
// feature: implement IPC transport
// feature: dds support get remote participant vbs version and rm some print
// ------------------------------------------------------------------

#include <rtps/participant/RTPSParticipantImpl.h>

#include <edds/rtps/attributes/ServerAttributes.h>
#include <edds/rtps/builtin/BuiltinProtocols.h>
#include <discovery/endpoint/EDP.h>
#include <discovery/participant/PDP.h>
#include <discovery/participant/PDPSimple.h>
#include <edds/rtps/builtin/data/ParticipantProxyData.h>
#include <edds/rtps/builtin/liveliness/WLP.h>
#include <history/WriterHistory.h>
#include <edds/rtps/messages/MessageReceiver.h>
#include <edds/rtps/participant/RTPSParticipant.h>
#include <edds/rtps/reader/StatelessReader.h>
#include <edds/rtps/reader/StatefulReader.h>
#include <edds/rtps/reader/StatelessPersistentReader.h>
#include <edds/rtps/reader/StatefulPersistentReader.h>
#include <edds/rtps/RTPSDomain.h>
#include <transport/UDSDgramTransportDescriptor.h>
#include <transport/UDPv4TransportDescriptor.h>
#include <transport/TransportFactory.hpp>
#include <transport/RAWTransportDescriptor.h>
#ifdef EDDS_TCP_TRANSPORT_COMPILE
#include <transport/TCPv4TransportDescriptor.h>
#endif
#ifdef VBS_IPV6
#include <transport/TCPv6TransportDescriptor.h>
#endif
#include <transport/SharedMemTransportDescriptor.h>
#include <edds/rtps/writer/StatelessWriter.h>
#include <edds/rtps/writer/StatefulWriter.h>
#include <edds/rtps/writer/StatelessPersistentWriter.h>
#include <edds/rtps/writer/StatefulPersistentWriter.h>

#include <deps/common/IPFinder.h>
#include <ertps/utils/Semaphore.h>
#include <ertps/xmlparser/XMLProfileManager.h>
//#include <SharedMemTransport.h>
#include <rtps/persistence/PersistenceService.h>

#include <algorithm>
#include <functional>
#include <memory>
#include <mutex>
#include <sstream>

#include <elog/Log.hpp>
#include <history/BasicPayloadPool.hpp>
#include <rtps/network/ExternalLocatorsProcessor.hpp>

namespace evbs {
namespace ertps {
namespace rtps {

using UDPv4TransportDescriptor = vbs::transport::UDPv4TransportDescriptor;
#ifdef EDDS_TCP_TRANSPORT_COMPILE
using TCPTransportDescriptor = vbs::transport::TCPTransportDescriptor;
#endif
using SharedMemTransportDescriptor = vbs::transport::SharedMemTransportDescriptor;
using UDPDgramTransportDescriptor = vbs::transport::UDSDgramTransportDescriptor;

inline std::string pid_to_hex_string(uint32_t num) {
    if (num == 0) {
        return "0";
    }

    const char* hexDigits = "0123456789ABCDEF";  // 十六进制字符
    char buffer[10];  // 8个十六进制数字（uint32_t） + 2个字符（0x）+ 1个空字符
    char* ptr = buffer + sizeof(buffer) - 1;  // 指向最后一个字符的位置
    *ptr = '\0';                              // 设置结尾字符

    while (num > 0) {
        *(--ptr) = hexDigits[num % 16];
        num /= 16;
    }

    return ptr;
}

inline std::string part_id_to_fixed_width_hex(uint32_t num) {
    const char* hexDigits = "0123456789ABCDEF";  // 十六进制字符
    char buffer[3];                              // 2个十六进制数字 + 1个空字符
    char* ptr = buffer + sizeof(buffer) - 1;     // 指向最后一个字符的位置
    *ptr = '\0';                                 // 设置结尾字符

    while (ptr != buffer) {
        *(--ptr) = hexDigits[num % 16];
        num /= 16;
    }

    return ptr;
}

static EntityId_t TrustedWriter(const EntityId_t& reader) {
    return (reader == c_EntityId_SPDPReader)              ? c_EntityId_SPDPWriter
           : (reader == c_EntityId_SEDPPubReader)         ? c_EntityId_SEDPPubWriter
           : (reader == c_EntityId_SEDPSubReader)         ? c_EntityId_SEDPSubWriter
           : (reader == c_EntityId_ReaderLiveliness)      ? c_EntityId_WriterLiveliness
           : (reader == c_EntityId_Compatible_SPDPReader) ? c_EntityId_SPDPWriter
                                                          : c_EntityId_Unknown;
}

static bool should_be_intraprocess_only(const RTPSParticipantAttributes& att) {
    return ((xmlparser::XMLProfileManager::library_settings().intraprocess_delivery == INTRAPROCESS_FULL) &&
            (att.builtin.discovery_config.ignoreParticipantFlags ==
             (ParticipantFilteringFlags::FILTER_DIFFERENT_HOST | ParticipantFilteringFlags::FILTER_DIFFERENT_PROCESS)));
}

static bool get_unique_flows_parameters(const RTPSParticipantAttributes& part_att, const EndpointAttributes& att,
                                        bool& unique_flows, uint16_t& initial_port, uint16_t& final_port) {
    const std::string* value = PropertyPolicyHelper::find_property(att.properties, "edds.unique_network_flows");

    unique_flows = (nullptr != value);
    if (unique_flows) {
        // TODO (Miguel C): parse value to get port range
        final_port = part_att.port.portBase;
        initial_port = part_att.port.portBase - 400U;
    }

    return true;
}

static uint16_t calculate_well_known_port(uint32_t domain_id, const RTPSParticipantAttributes& att, bool is_multicast) {
    uint32_t port =
        att.port.portBase + att.port.domainIDGain * domain_id +
        (is_multicast ? att.port.offsetd2
                      : (att.port.offsetd3 + att.port.participantIDGain * static_cast<uint32_t>(att.participantID)));
    if (port > 65535U) {
        elogError(RTPS_MSG_OUT, RetCode_t::RETCODE_OUT_OF_RESOURCES,  //LCOV_EXCL_START
                  "Calculated port number is too high. Probably the domainId is over 232, there are "
                      << "too much participants created or portBase is too high.");
        throw std::runtime_error("Calculated port number is too high. Probably the domainId is over 232");
        //exit(EXIT_FAILURE);
    }  //LCOV_EXCL_STOP

    return static_cast<uint16_t>(port);
}

Locator_t& RTPSParticipantImpl::applyLocatorAdaptRule(Locator_t& loc) {
    // This is a completely made up rule
    // It is transport responsibility to interpret this new port.
    loc.port += m_att.port.participantIDGain;
    return loc;
}

RTPSParticipantImpl::RTPSParticipantImpl(uint32_t domain_id, const RTPSParticipantAttributes& PParam,
                                         const GuidPrefix_t& guidP, const GuidPrefix_t& persistence_guid,
                                         RTPSParticipant* par, vbs::common::ParticipantListener* plisten)
    : has_shm_transport_(false),
      domain_id_(domain_id),
      m_att(PParam),
      m_guid(guidP, c_EntityId_RTPSParticipant),
      mp_builtinProtocols(nullptr),
      //   mp_ResourceSemaphore(new Semaphore(0U)),
      IdCounter(0U),
      type_check_fn_(nullptr),
      //   client_override_(false),
      internal_metatraffic_locators_(false),
      internal_default_locators_(false)
#if HAVE_SECURITY
      ,
      m_security_manager(this),
      m_auth_token()
#endif  // if HAVE_SECURITY
      ,
      mp_participantListener(plisten),
      mp_userParticipant(par),
      mp_mutex(new std::recursive_mutex()),
      is_intraprocess_only_(should_be_intraprocess_only(PParam)) {
    if (domain_id_ == edds::dds::DOMAIN_ID_UNKNOWN) {
        elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_BAD_PARAMETER, "Domain ID has to be set to a correct value");
        return;
    }
    if (c_GuidPrefix_Unknown != persistence_guid) {
        m_persistence_guid = GUID_t(persistence_guid, c_EntityId_RTPSParticipant);
    }

    mp_userParticipant->mp_impl = this;
    mp_event_thr.init_thread();

    m_transportFactory = vbs::transport::TransportFactory::get_instance();
    if (m_transportFactory == nullptr) {
        elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_ERROR,  //LCOV_EXCL_START
                  "The transport Factory instance is not correctly initialized");
        return;
    }  //LCOV_EXCL_STOP

    {
        const std::string* max_size_property =
            PropertyPolicyHelper::find_property(m_att.properties, "edds.max_message_size");
        if (max_size_property != nullptr) {
            try {
                max_output_message_size_ = std::stoul(*max_size_property);
            } catch (const std::exception& e)  //LCOV_EXCL_START
            {
                elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_ERROR,
                          "Error parsing max_message_size property: " << e.what());
            }
        }  //LCOV_EXCL_STOP
    }
#if HAVE_SECURITY
    // Start security
    if (!m_security_manager.init(security_attributes_, PParam.properties)) {
        // Participant will be deleted, no need to allocate buffers or create builtin endpoints
        return;
    }
#endif  // if HAVE_SECURITY

    bool bRegistered = false;
    if (m_att.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol_t::CLIENT ||
        m_att.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol_t::SERVER ||
        m_att.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol_t::BACKUP) {
#if !defined(_WIN32)
        bRegistered = RegisterTransport(vbs::transport::TransmitorType::TRANSMITOR_TYPE_UDS);
        if (!bRegistered) {
            elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_ERROR,  //LCOV_EXCL_START
                      "Participant " << getGuid() << " Register transport UDS faild");
            return;
        }  //LCOV_EXCL_STOP
        b_registed_transport = true;
        if (m_att.builtin.discovery_config.is_compatible &&
            (m_att.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol_t::SERVER ||
             m_att.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol_t::BACKUP)) {
            bRegistered = RegisterTransport(vbs::transport::TransmitorType::TRANSMITOR_TYPE_RAW);
            if (!bRegistered) {
                elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_ERROR,  //LCOV_EXCL_START
                          "Participant " << getGuid() << " Register transport RAW faild");
                return;
            }  //LCOV_EXCL_STOP
        }
#endif
        if (m_att.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol_t::CLIENT &&
            m_att.transport_type == ertps::rtps::TransportTypeKind::TRANSPORT_TYPE_UDP) {
            bRegistered = RegisterTransport(vbs::transport::TransmitorType::TRANSMITOR_TYPE_UDP);
            if (!bRegistered) {
                elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_ERROR,  //LCOV_EXCL_START
                          "Participant " << getGuid() << "RTPS Client register UDP transport faild");
                return;
            }  //LCOV_EXCL_STOP
        }
        if ((m_att.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol_t::SERVER ||
             m_att.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol_t::BACKUP) &&
            !m_att.userTransports.empty()) {
            bRegistered = RegisterTransport(vbs::transport::TransmitorType::TRANSMITOR_TYPE_MAX);
            if (!bRegistered) {
                elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_ERROR,  //LCOV_EXCL_START
                          "Participant " << getGuid() << "RTPS Server register transport faild");
                // return; // DServer can continue to operate even when no listening IP is available.
            }  //LCOV_EXCL_STOP
        }
    } else {
        if (!RegisterTransport(vbs::transport::TransmitorType::TRANSMITOR_TYPE_MAX)) {
            elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_ERROR,
                      "Participant " << getGuid() << "RTPS SIMPLE register transport faild");
            return;
        }
        b_registed_transport = true;
    }

#if 0
    namespace ExternalLocatorsProcessor = edds::rtps::ExternalLocatorsProcessor;
    ExternalLocatorsProcessor::set_listening_locators(m_att.builtin.metatraffic_external_unicast_locators,
                                                      m_att.builtin.metatrafficUnicastLocatorList);
    ExternalLocatorsProcessor::set_listening_locators(m_att.default_external_unicast_locators,
                                                      m_att.defaultUnicastLocatorList);
#endif

    bool allow_growing_buffers = m_att.allocation.send_buffers.dynamic;
    size_t num_send_buffers = m_att.allocation.send_buffers.preallocated_number;
    if (num_send_buffers == 0U) {
        // Two buffers (user, events)
        num_send_buffers = 2U;
        // Add one buffer per reception thread
        num_send_buffers += m_receiverResourcelist.size();
    }

    // Create buffer pool
    send_buffers_.reset(new SendBuffersManager(num_send_buffers, allow_growing_buffers));
    if (!send_buffers_->init(this)) {
        return;
    }

    // Initialize flow controller factory.
    // This must be done after initiate network layer.
    flow_controller_factory_.init(this);

    // Register user's flow controllers.
    for (auto& flow_controller_desc : m_att.flow_controllers) {
        flow_controller_factory_.register_flow_controller(*flow_controller_desc.get());
    }
    // 允许Client和SuperClient使用默认metatrafficUnicastLocatorList及m_DiscoveryServers
    get_default_metatraffic_locators();

#if HAVE_SECURITY
    if (m_security_manager.is_security_active()) {
        if (!m_security_manager.create_entities()) {
            return;
        }
    }
#endif  // if HAVE_SECURITY

    mp_builtinProtocols = new BuiltinProtocols();

    // Initialize builtin protocols
    if ((!mp_builtinProtocols->initBuiltinProtocols(this, m_att.builtin))) {
        elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_ERROR, "The builtin protocols were not correctly initialized");
        return;
    }

    if (c_GuidPrefix_Unknown != persistence_guid) {
        logDebug(RTPS_PARTICIPANT, "RTPSParticipant \"" << m_att.getName()
                                                        << "\" with guidPrefix: " << m_guid.guidPrefix
                                                        << " and persistence guid: " << persistence_guid);
    } else {
        logDebug(RTPS_PARTICIPANT,
                 "RTPSParticipant \"" << m_att.getName() << "\" with guidPrefix: " << m_guid.guidPrefix);
    }

    initialized_ = true;
}

void RTPSParticipantImpl::get_default_metatraffic_locators() {
    switch (m_att.builtin.discovery_config.discoveryProtocol) {
        case DiscoveryProtocol_t::NONE:
            logWarning(RTPS_PDP, "No participant discovery protocol specified");
            break;

        case DiscoveryProtocol_t::CLIENT:
        case DiscoveryProtocol_t::SUPER_CLIENT:
            // 假如CLIENT未配置则使用默认的UDS配置
            if (m_att.builtin.discovery_config.m_DiscoveryServers.empty()) {
                Locator_t locator;
                locator.kind = LOCATOR_KIND_UDS;
                constexpr size_t max_address_len = sizeof(locator.address);
                std::strncpy(reinterpret_cast<char*>(locator.address), "dserver", max_address_len - 1);
                locator.address[max_address_len - 1] = '\0';
                m_att.builtin.discovery_config.m_DiscoveryServers.push_back(locator);
            }

            if (m_att.builtin.metatrafficUnicastLocatorList.empty()) {
                Locator_t locator;
                locator.kind = LOCATOR_KIND_UDS;
                std::string path = pid_to_hex_string(getGuid().get_process_id()) +
                                   part_id_to_fixed_width_hex(getGuid().get_participant_id() % 0xFF);

                constexpr size_t max_address_len = sizeof(locator.address);
                std::strncpy(reinterpret_cast<char*>(locator.address), path.c_str(), max_address_len - 1);
                locator.address[max_address_len - 1] = '\0';
                m_att.builtin.metatrafficUnicastLocatorList.push_back(locator);
            }
            break;

        default:
            logDebug(RTPS_PDP, "DiscoveryProtocol_t " << m_att.builtin.discovery_config.discoveryProtocol
                                                      << " don't need get metatraffic_locators");
            break;
    }
}

RTPSParticipantImpl::RTPSParticipantImpl(uint32_t domain_id, const RTPSParticipantAttributes& PParam,
                                         const GuidPrefix_t& guidP, RTPSParticipant* par,
                                         vbs::common::ParticipantListener* plisten)
    : RTPSParticipantImpl(domain_id, PParam, guidP, c_GuidPrefix_Unknown, par, plisten) {}

bool RTPSParticipantImpl::RegisterTransport(vbs::transport::TransmitorType type) {
    bool bRegistered = false;

#if !defined(_WIN32)
    if (type == TransmitorType::TRANSMITOR_TYPE_UDS) {
        UDSDgramTransportDescriptor descriptor;
        descriptor.sendBufferSize = m_att.sendSocketBufferSize;
        descriptor.receiveBufferSize = m_att.listenSocketBufferSize;
        bRegistered = m_transportFactory->RegisterTransport(get_transport_name(), &descriptor);
        logDebug(RTPS_PARTICIPANT, "Participant " << getGuid() << "Register Transport UDS");
        if (!bRegistered) {
            elogInfo(RTPS_PARTICIPANT, "User transport failed to register.");
            return false;
        } else {
            if (descriptor.max_message_size() < m_maxMessageSize) {
                m_maxMessageSize = descriptor.max_message_size();
            }
            auto minSendBufferSize = m_transportFactory->GetMinSendBufferSize(
                get_transport_name(), vbs::transport::TransmitorType::TRANSMITOR_TYPE_UDS);
            if (minSendBufferSize < m_minSendBufferSize) {
                m_minSendBufferSize = minSendBufferSize;
            }
        }

        (void)createReceiverResources(m_att.builtin.metatrafficUnicastLocatorList, true, false, LOCATOR_KIND_UDS);
        (void)createSenderResources(m_att.builtin.metatrafficUnicastLocatorList);
        registed_user_transport(vbs::RemoteEndpointInfo::LocatorType::LOCATOR_TYPE_UDS, true);
        return true;
    } else if (type == TransmitorType::TRANSMITOR_TYPE_RAW) {
        RAWTransportDescriptor descriptor;
        descriptor.sendBufferSize = m_att.sendSocketBufferSize;
        descriptor.receiveBufferSize = m_att.listenSocketBufferSize;
        bRegistered = m_transportFactory->RegisterTransport(get_transport_name(), &descriptor);
        logDebug(RTPS_PARTICIPANT, "Participant " << getGuid() << "Register Transport RAW");
        if (!bRegistered) {
            elogInfo(RTPS_PARTICIPANT, "Builtin RAW transport failed to register.");
            return false;
        } else {
            if (descriptor.max_message_size() < m_maxMessageSize) {
                m_maxMessageSize = descriptor.max_message_size();
            }
            auto minSendBufferSize = m_transportFactory->GetMinSendBufferSize(
                get_transport_name(), vbs::transport::TransmitorType::TRANSMITOR_TYPE_RAW);
            if (minSendBufferSize < m_minSendBufferSize) {
                m_minSendBufferSize = minSendBufferSize;
            }
        }

        Locator_t multicast_locator;
        evbs::ertps::rtps::IPLocator::setIPv4(multicast_locator, "239.255.0.1");
        // evbs::ertps::rtps::IPLocator::setPhysicalPort(multicast_locator, 7400);
        multicast_locator.kind = LOCATOR_KIND_RAW;
        m_att.builtin.rawMulticastLocatorList.push_back(multicast_locator);

        (void)createReceiverResources(m_att.builtin.rawMulticastLocatorList, false, false, LOCATOR_KIND_RAW);
        return true;
    }
#endif

    if (m_att.useBuiltinTransports || ((m_att.useBuiltinTransports == false) && m_att.userTransports.empty())) {
        UDPv4TransportDescriptor descriptor;
        descriptor.sendBufferSize = m_att.sendSocketBufferSize;
        descriptor.receiveBufferSize = m_att.listenSocketBufferSize;
        bRegistered = m_transportFactory->RegisterTransport(get_transport_name(), &descriptor);
        logDebug(RTPS_PARTICIPANT, "Participant " << getGuid() << "Register Transport default");
        if (!bRegistered) {
            elogInfo(RTPS_PARTICIPANT, "User transport failed to register.");
        } else {
            if (descriptor.max_message_size() < m_maxMessageSize) {
                m_maxMessageSize = descriptor.max_message_size();
            }
            auto minSendBufferSize = m_transportFactory->GetMinSendBufferSize(
                get_transport_name(), vbs::transport::TransmitorType::TRANSMITOR_TYPE_UDP);
            if (minSendBufferSize < m_minSendBufferSize) {
                m_minSendBufferSize = minSendBufferSize;
            }
        }
    } else {
        // User defined transports
        for (const auto& transportDescriptor : m_att.userTransports) {
            if (type == vbs::transport::TransmitorType::TRANSMITOR_TYPE_MAX ||
                transportDescriptor.get()->get_type() == type) {
                bRegistered = m_transportFactory->RegisterTransport(get_transport_name(), transportDescriptor.get());
            }

            if (!bRegistered) {
                elogInfo(RTPS_PARTICIPANT, "User transport failed to register type:" << static_cast<uint32_t>(type));
            } else {
                logDebug(RTPS_PARTICIPANT, "Participant " << getGuid() << "Register user transport success.");

                if (transportDescriptor.get()->max_message_size() < m_maxMessageSize) {
                    m_maxMessageSize = transportDescriptor.get()->max_message_size();
                }
                auto minSendBufferSize = m_transportFactory->GetMinSendBufferSize(
                    get_transport_name(), transportDescriptor.get()->get_type());
                if (minSendBufferSize < m_minSendBufferSize) {
                    m_minSendBufferSize = minSendBufferSize;
                }
            }
        }
    }
    // Only when the user registers the transport flag as true, the subsequent filtering locator is used
    if (type == vbs::transport::TransmitorType::TRANSMITOR_TYPE_UDP) {
        registed_user_transport(vbs::RemoteEndpointInfo::LocatorType::LOCATOR_TYPE_UDP, true);
    }

    if (!bRegistered) {
        return false;
    }

    /* If metatrafficMulticastLocatorList is empty, add mandatory default Locators
       Else -> Take them */
    /* 用户创建udp transport不需要metatraffic */
    if (type == vbs::transport::TransmitorType::TRANSMITOR_TYPE_MAX) {
        // Creation of metatraffic locator and receiver resources
        uint32_t metatraffic_multicast_port = m_att.port.getMulticastPort(domain_id_);
        uint32_t metatraffic_unicast_port =
            m_att.port.getUnicastPort(domain_id_, static_cast<uint32_t>(m_att.participantID));
        uint32_t meta_multicast_port_for_check = metatraffic_multicast_port;

        /* INSERT DEFAULT MANDATORY MULTICAST LOCATORS HERE */
        if (m_att.builtin.metatrafficMulticastLocatorList.empty() &&
            m_att.builtin.metatrafficUnicastLocatorList.empty()) {
#if 0
            uint32_t metatraffic_multicast_port = m_att.port.getMulticastPort(domain_id_);
            uint32_t metatraffic_unicast_port =
                m_att.port.getUnicastPort(domain_id_, static_cast<uint32_t>(m_att.participantID));
#endif
            if (m_att.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol_t::SIMPLE ||
                m_att.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol_t::SERVER ||
                m_att.builtin.discovery_config.discoveryProtocol == DiscoveryProtocol_t::BACKUP) {
                (void)m_transportFactory->getDefaultMetatrafficMulticastLocators(
                    get_transport_name(), m_att.builtin.metatrafficMulticastLocatorList, metatraffic_multicast_port);
            }
            (void)m_transportFactory->getDefaultMetatrafficUnicastLocators(
                get_transport_name(), m_att.builtin.metatrafficUnicastLocatorList, metatraffic_unicast_port);
            internal_metatraffic_locators_ = true;
        } else {
            if ((0U < m_att.builtin.metatrafficMulticastLocatorList.size()) &&
                (0U != m_att.builtin.metatrafficMulticastLocatorList.begin()->port)) {
                meta_multicast_port_for_check = m_att.builtin.metatrafficMulticastLocatorList.begin()->port;
            }
            (void)std::for_each(m_att.builtin.metatrafficMulticastLocatorList.begin(),
                                m_att.builtin.metatrafficMulticastLocatorList.end(), [&](Locator_t& locator) {
                                    (void)m_transportFactory->fillMetatrafficMulticastLocator(
                                        get_transport_name(), locator, metatraffic_multicast_port);
                                });
            (void)std::for_each(m_att.builtin.metatrafficUnicastLocatorList.begin(),
                                m_att.builtin.metatrafficUnicastLocatorList.end(), [&](Locator_t& locator) {
                                    (void)m_transportFactory->fillMetatrafficUnicastLocator(
                                        get_transport_name(), locator, metatraffic_unicast_port);
                                });
        }
        m_transportFactory->NormalizeLocators(get_transport_name(), m_att.builtin.metatrafficMulticastLocatorList);
        m_transportFactory->NormalizeLocators(get_transport_name(), m_att.builtin.metatrafficUnicastLocatorList);
        // Check metatraffic multicast port
        if ((0U < m_att.builtin.metatrafficMulticastLocatorList.size()) &&
            (m_att.builtin.metatrafficMulticastLocatorList.begin()->port != meta_multicast_port_for_check)) {
            logWarning(RTPS_PARTICIPANT, "Metatraffic multicast port "
                                             << meta_multicast_port_for_check
                                             << " cannot be opened."
                                                " It may is opened by another application. Discovery may fail.");
        }

        // Initial peers
        if (m_att.builtin.initialPeersList.empty()) {
            m_att.builtin.initialPeersList = m_att.builtin.metatrafficMulticastLocatorList;
        } else {
            LocatorList_t initial_peers;
            initial_peers.swap(m_att.builtin.initialPeersList);

            (void)std::for_each(initial_peers.begin(), initial_peers.end(), [&](Locator_t& locator) {
                if (locator.port == 0) {
                    for (uint32_t i = 0; i < s_maximumInitialPeersRange; ++i) {
                        Locator auxloc(locator);
                        auxloc.port = m_att.port.getUnicastPort(domain_id_, i);

                        m_att.builtin.initialPeersList.push_back(auxloc);
                    }
                } else {
                    m_att.builtin.initialPeersList.push_back(locator);
                }
            });
        }
    }
    // Creation of user locator and receiver resources
    // If no default locators are defined we define some.
    /* The reasoning here is the following.
       If the parameters of the RTPS Participant don't hold default listening locators for the
       creation of Endpoints, we make some for Unicast only. If there is at least one listen locator
       of any kind, we do not create any default ones. If there are no sending locators defined, we
       create default ones for the transports we implement.
     */
    if (m_att.defaultUnicastLocatorList.empty() && m_att.defaultMulticastLocatorList.empty()) {
        // Default Unicast Locators in case they have not been provided
        /* INSERT DEFAULT UNICAST LOCATORS FOR THE PARTICIPANT */

        uint16_t unicast_port = calculate_well_known_port(domain_id_, m_att, false);

        (void)m_transportFactory->getDefaultUnicastLocators(get_transport_name(), m_att.defaultUnicastLocatorList,
                                                            unicast_port);
        m_transportFactory->NormalizeLocators(get_transport_name(), m_att.defaultUnicastLocatorList);

        internal_default_locators_ = true;
        logDebug(RTPS_PARTICIPANT, m_att.getName() << " Created with NO default Unicast Locator List, adding Locators:"
                                                   << m_att.defaultUnicastLocatorList);
    } else {
        // Locator with port 0, calculate port.
        uint16_t unicast_port = calculate_well_known_port(domain_id_, m_att, false);
        for (auto& loc : m_att.defaultUnicastLocatorList) {
            (void)m_transportFactory->fill_default_locator_port(get_transport_name(), loc, unicast_port);
        }
        uint16_t multicast_port = calculate_well_known_port(domain_id_, m_att, true);
        for (auto& loc : m_att.defaultMulticastLocatorList) {
            (void)m_transportFactory->fill_default_locator_port(get_transport_name(), loc, multicast_port);
        }
    }

    /* 用户创建udp transport不需要metatraffic */
    if (type == vbs::transport::TransmitorType::TRANSMITOR_TYPE_MAX) {
        (void)createReceiverResources(m_att.builtin.metatrafficMulticastLocatorList, true, initialized_);
        (void)createReceiverResources(m_att.builtin.metatrafficUnicastLocatorList, true, initialized_);
    }

    (void)createReceiverResources(m_att.defaultUnicastLocatorList, true, initialized_);
    (void)createReceiverResources(m_att.defaultMulticastLocatorList, true, initialized_);
    (void)createSenderResources(m_att.defaultUnicastLocatorList);

    return true;
}

bool RTPSParticipantImpl::bTransportRegistered(TransmitorType type) {
    return m_transportFactory->bTransportRegistered(get_transport_name(), type);
}

void RTPSParticipantImpl::enable() {
#ifdef EDDS_SHM_TRANSPORT_COMPILE
#if defined(__linux)
    if (has_shm_transport_) {
        SharedMemTransportDescriptor shm_transport;
        std::unique_ptr<TransportInterface> transport(shm_transport.create_transport());
        if (transport) {
            (void)transport->init();
            transport->shared_mem_clean();
        }
    }
#endif
#endif
    mp_builtinProtocols->enable();

    // Start reception
    for (auto& receiver : m_receiverResourcelist) {
        receiver.Receiver->RegisterReceiver(receiver.mp_receiver);
    }
    mp_builtinProtocols->enable_second();
}

void RTPSParticipantImpl::disable() {
    // Disabling event thread also disables participant announcement, so there is no need to call
    // stopRTPSParticipantAnnouncement()
    mp_event_thr.stop_thread();

    if (!m_transportFactory || !b_registed_transport) {
        return;
    }
    // Disable Retries on Transports
    m_transportFactory->Shutdown(get_transport_name());

    // Safely abort threads.
    for (auto& block : m_receiverResourcelist) {
        block.Receiver->UnregisterReceiver(block.mp_receiver);
        block.disable();
    }

    deleteAllUserEndpoints();

    if (nullptr != mp_builtinProtocols) {
        delete (mp_builtinProtocols);
        mp_builtinProtocols = nullptr;
    }
}

const std::vector<RTPSWriter*>& RTPSParticipantImpl::getAllWriters() const {
    return m_allWriterList;
}

const std::vector<RTPSReader*>& RTPSParticipantImpl::getAllReaders() const {
    return m_allReaderList;
}

RTPSParticipantImpl::~RTPSParticipantImpl() {
    disable();

#if HAVE_SECURITY
    m_security_manager.destroy();
#endif  // if HAVE_SECURITY

    // Destruct message receivers
    for (auto& block : m_receiverResourcelist) {
        delete block.mp_receiver;
    }
    m_receiverResourcelist.clear();
    if (m_transportFactory && b_registed_transport) {
        m_transportFactory->UnregisterTransport(get_transport_name());
        m_transportFactory = nullptr;
    }
    // delete mp_ResourceSemaphore;
    delete mp_userParticipant;
    mp_userParticipant = nullptr;

    delete mp_mutex;
}

template <EndpointKind_t kind, octet no_key, octet with_key>
bool RTPSParticipantImpl::preprocess_endpoint_attributes(const EntityId_t& entity_id, std::atomic<uint32_t>& id_counter,
                                                         EndpointAttributes& att, EntityId_t& entId) {
    const char* debug_label = (att.endpointKind == WRITER ? "writer" : "reader");

    if (!att.unicastLocatorList.isValid()) {
        elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_BAD_PARAMETER,
                  "Unicast Locator List for " << debug_label << " contains invalid Locator");
        return false;
    }
    if (!att.multicastLocatorList.isValid()) {
        elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_BAD_PARAMETER,
                  "Multicast Locator List for " << debug_label << " contains invalid Locator");
        return false;
    }
    if (!att.remoteLocatorList.isValid()) {
        elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_BAD_PARAMETER,
                  "Remote Locator List for " << debug_label << " contains invalid Locator");
        return false;
    }

    if (entity_id == c_EntityId_Unknown) {
        if (att.topicKind == NO_KEY) {
            entId.value[3] = ((-2 == att.getUserDefinedID()) && (0 < att.getEntityID())) ? (0x60U) | no_key : no_key;
        } else if (att.topicKind == WITH_KEY) {
            entId.value[3] = ((-2 == att.getUserDefinedID()) && (0 < att.getEntityID())) ? (0x60) | with_key : with_key;
        }
        uint32_t idnum;
        if (att.getEntityID() > 0) {
            idnum = static_cast<uint32_t>(att.getEntityID());
        } else {
            idnum = ++id_counter;
        }

        entId.value[2] = octet(idnum);
        entId.value[1] = octet(idnum >> 8);
        entId.value[0] = octet(idnum >> 16);
    } else {
        entId = entity_id;
    }
    if (att.persistence_guid == c_Guid_Unknown) {
        // Try to load persistence_guid from property
        const std::string* persistence_guid_property =
            PropertyPolicyHelper::find_property(att.properties, "dds.persistence.guid");
        if (persistence_guid_property != nullptr) {
            // Load persistence_guid from property
            std::istringstream(persistence_guid_property->c_str()) >> att.persistence_guid;
            if (att.persistence_guid == c_Guid_Unknown) {
                // Wrongly configured property
                elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_BAD_PARAMETER,
                          "Cannot configure " << debug_label << "'s persistence GUID from '"
                                              << persistence_guid_property->c_str() << "'. Wrong input");
                return false;
            }
        }
    }

    // Error log level can be disable. Avoid unused warning
    static_cast<void>(debug_label);

    return true;
}

template <typename Functor>
bool RTPSParticipantImpl::create_writer(RTPSWriter** writer_out, WriterAttributes& param, const EntityId_t& entity_id,
                                        bool is_builtin, const Functor& callback) {
    std::string type = (param.endpoint.reliabilityKind == RELIABLE) ? "RELIABLE" : "BEST_EFFORT";
    EntityId_t entId;

    if (!preprocess_endpoint_attributes<WRITER, 0x03, 0x02>(entity_id, IdCounter, param.endpoint, entId)) {
        logInfo(RTPS_PARTICIPANT, "preprocess entity id err. in entity:" << entity_id);
        return false;
    }

    if (existsEntityId(entId, WRITER)) {
        elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_PRECONDITION_NOT_MET,
                  "A writer with the same entityId already exists in this RTPSParticipant");
        return false;
    }

    GUID_t guid(m_guid.guidPrefix, entId);
    edds::rtps::FlowController* flow_controller = nullptr;
    const char* flow_controller_name = param.flow_controller_name;

    // Retrieve flow controller.
    // If not default flow controller, publish_mode must be asynchronously.
    if ((nullptr == flow_controller) &&
        ((vbsutil::flowctlString::EDDS_FLOW_CONTROLLER_DEFAULT == flow_controller_name) ||
         (ASYNCHRONOUS_WRITER == param.mode))) {
        flow_controller = flow_controller_factory_.retrieve_flow_controller(flow_controller_name, param);
    }

    if (nullptr == flow_controller) {
        if ((vbsutil::flowctlString::EDDS_FLOW_CONTROLLER_DEFAULT != flow_controller_name) &&
            (SYNCHRONOUS_WRITER == param.mode)) {
            elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                      "Cannot use a flow controller in synchronously publication mode.");
        } else {
            elogError(
                RTPS_PARTICIPANT, RetCode_t::RETCODE_PRECONDITION_NOT_MET,
                "Cannot create the writer. Couldn't find flow controller " << flow_controller_name << " for writer.");
        }
        return false;
    }

    // Check for unique_network_flows feature
    if (nullptr != PropertyPolicyHelper::find_property(param.endpoint.properties, "edds.unique_network_flows")) {
        elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_UNSUPPORTED, "Unique network flows not supported on writers");
        return false;
    }

    if (param.endpoint.persistence_guid == c_Guid_Unknown) {
        if (m_persistence_guid != c_Guid_Unknown) {
            // Generate persistence guid from participant persistence guid
            param.endpoint.persistence_guid = GUID_t(m_persistence_guid.guidPrefix, entity_id);
        }
    }

    // Get persistence service
    IPersistenceService* persistence = nullptr;
    if (!get_persistence_service(is_builtin, param.endpoint, persistence)) {
        return false;
    }

    normalize_endpoint_locators(param.endpoint);

    RTPSWriter* SWriter = nullptr;
    SWriter = callback(guid, param, flow_controller, persistence, param.endpoint.reliabilityKind == RELIABLE);

    if (SWriter == nullptr) {
        return false;
    }

    if (!SWriter->is_pool_initialized()) {
        delete (SWriter);
        return false;
    }

#if HAVE_SECURITY
    if (!is_builtin) {
        if (!m_security_manager.register_local_writer(SWriter->getGuid(), param.endpoint.properties,
                                                      SWriter->getAttributes().security_attributes())) {
            delete (SWriter);
            return false;
        }
    } else {
        if (!m_security_manager.register_local_builtin_writer(SWriter->getGuid(),
                                                              SWriter->getAttributes().security_attributes())) {
            delete (SWriter);
            return false;
        }
    }
#endif  // if HAVE_SECURITY

    //create transport resources of user-endpoints later, if transport type is UDP, create send resources and
    //associate receivers with endpoint
    if (is_builtin || m_att.transport_type == ertps::rtps::TransportTypeKind::TRANSPORT_TYPE_UDP) {
        if (is_builtin) {
            SWriter->builtin = true;
        }

        (void)createSendResources(SWriter);

        if (param.endpoint.reliabilityKind == RELIABLE) {
            if (!createAndAssociateReceiverswithEndpoint(SWriter)) {
                delete (SWriter);
                return false;
            }
        }
    }

    {
        std::lock_guard<vbs::shared_mutex> _(endpoints_list_mutex);
        m_allWriterList.push_back(SWriter);

        if (!is_builtin) {
            m_userWriterList.push_back(SWriter);
        }
    }

    *writer_out = SWriter;

    logDebug(RTPS_PARTICIPANT, "Creating writer of type " << type << " Guid:" << guid);
    return true;
}

template <typename Functor>
bool RTPSParticipantImpl::create_reader(RTPSReader** reader_out, ReaderAttributes& param, const EntityId_t& entity_id,
                                        bool is_builtin, bool enable, const Functor& callback) {
    std::string type = (param.endpoint.reliabilityKind == RELIABLE) ? "RELIABLE" : "BEST_EFFORT";
    EntityId_t entId;

    if (!preprocess_endpoint_attributes<READER, 0x04, 0x07>(entity_id, IdCounter, param.endpoint, entId)) {
        logInfo(RTPS_PARTICIPANT, "preprocess entity id err. in entity:" << entity_id);
        return false;
    }

    if (existsEntityId(entId, READER)) {
        elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_PRECONDITION_NOT_MET,
                  "A reader with the same entityId already exists in this RTPSParticipant");
        return false;
    }

    if (param.endpoint.persistence_guid == c_Guid_Unknown) {
        if (m_persistence_guid != c_Guid_Unknown) {
            // Generate persistence guid from participant persistence guid
            param.endpoint.persistence_guid = GUID_t(m_persistence_guid.guidPrefix, entity_id);
        }
    }

    // Get persistence service
    IPersistenceService* persistence = nullptr;
    if (!get_persistence_service(is_builtin, param.endpoint, persistence)) {
        return false;
    }

    // Check for unique_network_flows feature
    bool request_unique_flows = false;
    uint16_t initial_port = 0U;
    uint16_t final_port = 0U;
    if (!get_unique_flows_parameters(m_att, param.endpoint, request_unique_flows, initial_port, final_port)) {
        return false;
    }

    normalize_endpoint_locators(param.endpoint);

    RTPSReader* SReader = nullptr;
    GUID_t guid(m_guid.guidPrefix, entId);
    SReader = callback(guid, param, persistence, param.endpoint.reliabilityKind == RELIABLE);

    if (SReader == nullptr) {
        return false;
    }

#if HAVE_SECURITY

    if (!is_builtin) {
        if (!m_security_manager.register_local_reader(SReader->getGuid(), param.endpoint.properties,
                                                      SReader->getAttributes().security_attributes())) {
            delete (SReader);
            return false;
        }
    } else {
        if (!m_security_manager.register_local_builtin_reader(SReader->getGuid(),
                                                              SReader->getAttributes().security_attributes())) {
            delete (SReader);
            return false;
        }
    }
#endif  // if HAVE_SECURITY

    //create transport resources of user-endpoints later
    if (is_builtin || m_att.transport_type == ertps::rtps::TransportTypeKind::TRANSPORT_TYPE_UDP) {
        if (is_builtin) {
            SReader->builtin = true;
        }

        if (param.endpoint.reliabilityKind == RELIABLE) {
            (void)createSendResources(SReader);
        }
        SReader->setTrustedWriter(TrustedWriter(SReader->getGuid().entityId));
        logDebug(RTPS_PARTICIPANT, "createAndAssociateReceiverswithEndpoint READER:" << SReader->getGuid());
        if (enable) {
            if (!createAndAssociateReceiverswithEndpoint(SReader, request_unique_flows, initial_port, final_port)) {
                delete (SReader);
                return false;
            }
        }
    }

    {
        std::lock_guard<vbs::shared_mutex> _(endpoints_list_mutex);

        m_allReaderList.push_back(SReader);

        if (!is_builtin) {
            m_userReaderList.push_back(SReader);
        }
    }

    *reader_out = SReader;
    logDebug(RTPS_PARTICIPANT, "Creating reader of type " << type << " Guid:" << guid);
    return true;
}

/*
 *
 * MAIN RTPSParticipant IMPL API
 *
 */
bool RTPSParticipantImpl::createWriter(RTPSWriter** WriterOut, WriterAttributes& param_w, WriterHistory* hist,
                                       WriterListener* listen, const EntityId_t& entityId, bool isBuiltin) {
    auto callback = [hist, listen, this](const GUID_t& guid, WriterAttributes& param,
                                         edds::rtps::FlowController* flow_controller, IPersistenceService* persistence,
                                         bool is_reliable) -> RTPSWriter* {
        if (is_reliable) {
            if (persistence != nullptr) {
                return new StatefulPersistentWriter(this, guid, param, flow_controller, hist, listen, persistence);
            } else {
                return new StatefulWriter(this, guid, param, flow_controller, hist, listen);
            }
        } else {
            if (persistence != nullptr) {
                return new StatelessPersistentWriter(this, guid, param, flow_controller, hist, listen, persistence);
            } else {
                return new StatelessWriter(this, guid, param, flow_controller, hist, listen);
            }
        }
    };
    bool ret;

    ret = create_writer(WriterOut, param_w, entityId, isBuiltin, callback);
    if (ret) {
        hist->guid = (*WriterOut)->getGuid();
        hist->topicName = (*WriterOut)->getTopicName();
        if (isBuiltin) {
            hist->notify_writer = std::bind(&RTPSWriter::unsent_change_added_to_history, *WriterOut,
                                            std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
        }
        hist->change_removed_by_history =
            std::bind(&RTPSWriter::change_removed_by_history, *WriterOut, std::placeholders::_1);
    }
    return ret;
}

bool RTPSParticipantImpl::createWriter(RTPSWriter** WriterOut, WriterAttributes& param,
                                       const std::shared_ptr<IPayloadPool>& payload_pool, WriterHistory* hist,
                                       WriterListener* listen, const EntityId_t& entityId, bool isBuiltin) {
    if (!payload_pool) {
        elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_BAD_PARAMETER, "Trying to create writer with null payload pool");
        return false;
    }

    auto callback = [hist, listen, &payload_pool, this](
                        const GUID_t& guid, WriterAttributes& param_w, edds::rtps::FlowController* flow_controller,
                        IPersistenceService* persistence, bool is_reliable) -> RTPSWriter* {
        if (is_reliable) {
            if (persistence != nullptr) {
                return new StatefulPersistentWriter(this, guid, param_w, payload_pool, flow_controller, hist, listen,
                                                    persistence);
            } else {
                return new StatefulWriter(this, guid, param_w, payload_pool, flow_controller, hist, listen);
            }
        } else {
            if (persistence != nullptr) {
                return new StatelessPersistentWriter(this, guid, param_w, payload_pool, flow_controller, hist, listen,
                                                     persistence);
            } else {
                return new StatelessWriter(this, guid, param_w, payload_pool, flow_controller, hist, listen);
            }
        }
    };
    bool ret;

    ret = create_writer(WriterOut, param, entityId, isBuiltin, callback);
    if (ret) {
        hist->guid = (*WriterOut)->getGuid();
        hist->topicName = (*WriterOut)->getTopicName();
        if (isBuiltin) {
            hist->notify_writer = std::bind(&RTPSWriter::unsent_change_added_to_history, *WriterOut,
                                            std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
        }
        hist->change_removed_by_history =
            std::bind(&RTPSWriter::change_removed_by_history, *WriterOut, std::placeholders::_1);
    }
    return ret;
}

bool RTPSParticipantImpl::createWriter(RTPSWriter** WriterOut, WriterAttributes& watt_w,
                                       const std::shared_ptr<IPayloadPool>& payload_pool,
                                       const std::shared_ptr<IChangePool>& change_pool, WriterHistory* hist,
                                       WriterListener* listen, const EntityId_t& entityId, bool isBuiltin) {
    if (!payload_pool) {
        elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_BAD_PARAMETER, "Trying to create writer with null payload pool");
        return false;
    }

    auto callback = [hist, listen, &payload_pool, &change_pool, this](
                        const GUID_t& guid, WriterAttributes& watt, edds::rtps::FlowController* flow_controller,
                        IPersistenceService* persistence, bool is_reliable) -> RTPSWriter* {
        if (is_reliable) {
            if (persistence != nullptr) {
                return new StatefulPersistentWriter(this, guid, watt, payload_pool, change_pool, flow_controller, hist,
                                                    listen, persistence);
            } else {
                return new StatefulWriter(this, guid, watt, payload_pool, change_pool, flow_controller, hist, listen);
            }
        } else {
            if (persistence != nullptr) {
                return new StatelessPersistentWriter(this, guid, watt, payload_pool, change_pool, flow_controller, hist,
                                                     listen, persistence);
            } else {
                return new StatelessWriter(this, guid, watt, payload_pool, change_pool, flow_controller, hist, listen);
            }
        }
    };
    return create_writer(WriterOut, watt_w, entityId, isBuiltin, callback);
}

bool RTPSParticipantImpl::createReader(RTPSReader** ReaderOut, ReaderAttributes& param, ReaderHistory* hist,
                                       ReaderListener* listen, const EntityId_t& entityId, bool isBuiltin,
                                       bool enable) {
    auto callback = [hist, listen, this](const GUID_t& guid, ReaderAttributes& param_r,
                                         IPersistenceService* persistence, bool is_reliable) -> RTPSReader* {
        try {
            if (is_reliable) {
                if (persistence != nullptr) {
                    return new StatefulPersistentReader(this, guid, param_r, hist, listen, persistence);
                } else {
                    return new StatefulReader(this, guid, param_r, hist, listen);
                }
            } else {
                if (persistence != nullptr) {
                    return new StatelessPersistentReader(this, guid, param_r, hist, listen, persistence);
                } else {
                    return new StatelessReader(this, guid, param_r, hist, listen);
                }
            }
        } catch (const foonathan::memory::out_of_memory& e) {  //LCOV_EXCL_START
            elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_OUT_OF_RESOURCES,
                      "new StatefulPersistentReader() catch err: " << e.what());
            return nullptr;
        }  //LCOV_EXCL_STOP
    };
    bool ret;

    ret = create_reader(ReaderOut, param, entityId, isBuiltin, enable, callback);
    if (ret) {
        hist->topicName = (*ReaderOut)->getTopicName();
        if (param.endpoint.reliabilityKind == RELIABLE) {
            hist->change_removed_by_history =
                std::bind(&StatefulReader::change_removed_by_history, dynamic_cast<StatefulReader*>(*ReaderOut),
                          std::placeholders::_1);
        }
    }
    return ret;
}

bool RTPSParticipantImpl::createReader(RTPSReader** ReaderOut, ReaderAttributes& param_r,
                                       const std::shared_ptr<IPayloadPool>& payload_pool, ReaderHistory* hist,
                                       ReaderListener* listen, const EntityId_t& entityId, bool isBuiltin,
                                       bool enable) {
    if (!payload_pool) {
        elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_BAD_PARAMETER, "Trying to create reader with null payload pool");
        return false;
    }

    auto callback = [hist, listen, &payload_pool, this](const GUID_t& guid, ReaderAttributes& param,
                                                        IPersistenceService* persistence,
                                                        bool is_reliable) -> RTPSReader* {
        try {
            if (is_reliable) {
                if (persistence != nullptr) {
                    return new StatefulPersistentReader(this, guid, param, payload_pool, hist, listen, persistence);
                } else {
                    return new StatefulReader(this, guid, param, payload_pool, hist, listen);
                }
            } else {
                if (persistence != nullptr) {
                    return new StatelessPersistentReader(this, guid, param, payload_pool, hist, listen, persistence);
                } else {
                    return new StatelessReader(this, guid, param, payload_pool, hist, listen);
                }
            }
        } catch (const foonathan::memory::out_of_memory& e) {  //LCOV_EXCL_START
            elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_OUT_OF_RESOURCES,
                      "new StatefulPersistentReader() catch err: " << e.what());
            return nullptr;
        }  //LCOV_EXCL_STOP
    };
    bool ret;

    ret = create_reader(ReaderOut, param_r, entityId, isBuiltin, enable, callback);
    if (ret) {
        hist->topicName = (*ReaderOut)->getTopicName();
        if (param_r.endpoint.reliabilityKind == RELIABLE) {
            hist->change_removed_by_history =
                std::bind(&StatefulReader::change_removed_by_history, dynamic_cast<StatefulReader*>(*ReaderOut),
                          std::placeholders::_1);
        }
        //hist->payload_pool_ = payload_pool;
    }
    return ret;
}

bool RTPSParticipantImpl::createReader(RTPSReader** ReaderOut, ReaderAttributes& param_r,
                                       const std::shared_ptr<IChangePool>& change_pool, ReaderHistory* hist,
                                       ReaderListener* listen, const EntityId_t& entityId, bool isBuiltin,
                                       bool enable) {
    if (!change_pool) {
        elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_BAD_PARAMETER,  //LCOV_EXCL_START
                  "Trying to create reader with null payload pool");
        return false;
    }  //LCOV_EXCL_STOP

    auto callback = [hist, listen, &change_pool, this](const GUID_t& guid, ReaderAttributes& param,
                                                       IPersistenceService* persistence,
                                                       bool is_reliable) -> RTPSReader* {
        try {
            if (is_reliable) {
                if (persistence != nullptr) {
                    return new StatefulPersistentReader(this, guid, param, change_pool, hist, listen, persistence);
                } else {
                    return new StatefulReader(this, guid, param, change_pool, hist, listen);
                }
            } else {
                if (persistence != nullptr) {
                    return new StatelessPersistentReader(this, guid, param, change_pool, hist, listen, persistence);
                } else {
                    return new StatelessReader(this, guid, param, change_pool, hist, listen);
                }
            }
        } catch (const foonathan::memory::out_of_memory& e) {  //LCOV_EXCL_START
            elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_OUT_OF_RESOURCES,
                      "new StatefulPersistentReader() catch err: " << e.what());
            return nullptr;
        }  //LCOV_EXCL_STOP
    };
    bool ret;

    ret = create_reader(ReaderOut, param_r, entityId, isBuiltin, enable, callback);
    if (ret) {
        hist->topicName = (*ReaderOut)->getTopicName();
        if (param_r.endpoint.reliabilityKind == RELIABLE) {
            hist->change_removed_by_history =
                std::bind(&StatefulReader::change_removed_by_history, dynamic_cast<StatefulReader*>(*ReaderOut),
                          std::placeholders::_1);
        }
        //hist->payload_pool_ = payload_pool;
    }
    return ret;
}

RTPSReader* RTPSParticipantImpl::find_local_reader(const GUID_t& reader_guid) {
    vbs::shared_lock<vbs::shared_mutex> _(endpoints_list_mutex);

    for (auto& reader : m_allReaderList) {
        if (reader->getGuid() == reader_guid) {
            return reader;
        }
    }

    return nullptr;
}

RTPSWriter* RTPSParticipantImpl::find_local_writer(const GUID_t& writer_guid) {
    vbs::shared_lock<vbs::shared_mutex> _(endpoints_list_mutex);

    for (auto& writer : m_allWriterList) {
        if (writer->getGuid() == writer_guid) {
            return writer;
        }
    }

    return nullptr;
}

bool RTPSParticipantImpl::enableReader(RTPSReader* reader) {
    if (!assignEndpointListenResources(reader)) {
        return false;
    }
    return true;
}

// Avoid to receive PDPSimple reader a DATA while calling ~PDPSimple and EDP was destroy already.
void RTPSParticipantImpl::disableReader(RTPSReader* reader) {
    vbs::shared_lock<vbs::shared_mutex> guard(m_receiverResourcelistMutex);
    for (auto it = m_receiverResourcelist.begin(); it != m_receiverResourcelist.end(); ++it) {
        it->mp_receiver->removeEndpoint(reader);
    }
}

bool RTPSParticipantImpl::registerWriter(RTPSWriter* Writer, const TopicAttributes& topicAtt, const WriterQos& wqos,
                                         const vbsutil::xmlparser::ContentFilterProperty* content_filter) {
    return this->mp_builtinProtocols->addLocalWriter(Writer, topicAtt, wqos, content_filter);
}

bool RTPSParticipantImpl::registerReader(RTPSReader* reader, const TopicAttributes& topicAtt, const ReaderQos& rqos,
                                         const vbsutil::xmlparser::ContentFilterProperty* content_filter) {
    return this->mp_builtinProtocols->addLocalReader(reader, topicAtt, rqos, content_filter);
}

void RTPSParticipantImpl::update_attributes(const RTPSParticipantAttributes& patt) {
    UNUSED_PARAMETER(patt);
#if 0
    bool local_interfaces_changed = false;

    // Check if new interfaces have been added
    if (internal_metatraffic_locators_) {
        LocatorList_t metatraffic_unicast_locator_list = m_att.builtin.metatrafficUnicastLocatorList;
        get_default_metatraffic_locators();
        if (!(metatraffic_unicast_locator_list == m_att.builtin.metatrafficUnicastLocatorList)) {
            local_interfaces_changed = true;
            logDebug(RTPS_PARTICIPANT, m_att.getName() << " updated its metatraffic locators , current locators:"
                                                      << m_att.builtin.metatrafficUnicastLocatorList);
        }
    }
    if (internal_default_locators_) {
        LocatorList_t default_unicast_locator_list = m_att.defaultUnicastLocatorList;
        get_default_unicast_locators();
        if (!(default_unicast_locator_list == m_att.defaultUnicastLocatorList)) {
            local_interfaces_changed = true;
            logDebug(RTPS_PARTICIPANT, m_att.getName() << " updated default unicast locator list, current locators: "
                                                      << m_att.defaultUnicastLocatorList);
        }
    }

    if (local_interfaces_changed) {
        m_transportFactory->update_network_interfaces(get_transport_name());
    }

    auto pdp        = mp_builtinProtocols->mp_PDP;
    bool update_pdp = false;

    // Check if there are changes
    if (local_interfaces_changed) {
        update_pdp = true;
        std::vector<GUID_t> modified_servers;
        LocatorList_t modified_locators;

        // Update RTPSParticipantAttributes members

        // If there's no PDP don't process Discovery-related attributes.
        if (!pdp) {
            return;
        }

        // Update listening locators on external locators
        {
            namespace ExternalLocatorsProcessor = edds::rtps::ExternalLocatorsProcessor;
            if (local_interfaces_changed && internal_metatraffic_locators_) {
                ExternalLocatorsProcessor::set_listening_locators(m_att.builtin.metatraffic_external_unicast_locators,
                                                                  m_att.builtin.metatrafficUnicastLocatorList);
            }
            if (local_interfaces_changed && internal_default_locators_) {
                ExternalLocatorsProcessor::set_listening_locators(m_att.default_external_unicast_locators,
                                                                  m_att.defaultUnicastLocatorList);
            }
        }

        {
            std::lock_guard<std::recursive_mutex> lock(*pdp->getMutex());

            // Update user data
            auto local_participant_proxy_data = pdp->getLocalParticipantProxyData();

            // Update metatraffic locators
            for (auto& locator : m_att.builtin.metatrafficMulticastLocatorList) {
                local_participant_proxy_data->metatraffic_locators.add_multicast_locator(locator);
            }
            for (auto& locator : m_att.builtin.metatrafficUnicastLocatorList) {
                local_participant_proxy_data->metatraffic_locators.add_unicast_locator(locator);
            }

            // Update default locators
            for (auto& locator : m_att.defaultUnicastLocatorList) {
                local_participant_proxy_data->default_locators.add_unicast_locator(locator);
            }

            createSenderResources(m_att.builtin.metatrafficMulticastLocatorList);
            createSenderResources(m_att.builtin.metatrafficUnicastLocatorList);
            createSenderResources(m_att.defaultUnicastLocatorList);
            if (!modified_locators.empty()) {
                createSenderResources(modified_locators);
            }
        }
    }

    if (update_pdp) {
        // Send DATA(P)
        pdp->announceParticipantState(true);
    }
#endif
}

bool RTPSParticipantImpl::updateLocalWriter(RTPSWriter* writer, const TopicAttributes& topicAtt, const WriterQos& wqos,
                                            const vbsutil::xmlparser::ContentFilterProperty* content_filter,
                                            bool notify) {
    //create transport resources of user-endpoints later
    (void)createSendResources(writer);
    if (wqos.m_reliability.kind == RELIABLE_RELIABILITY_QOS) {
        if (!createAndAssociateReceiverswithEndpoint(writer)) {
            logWarning(RTPS_PARTICIPANT, "create receiver endpoint failed when update local writer.");
        }
    }

    return this->mp_builtinProtocols->updateLocalWriter(writer, topicAtt, wqos, content_filter, notify);
}

bool RTPSParticipantImpl::updateLocalReader(RTPSReader* reader, const TopicAttributes& topicAtt, const ReaderQos& rqos,
                                            const vbsutil::xmlparser::ContentFilterProperty* content_filter,
                                            bool notify) {
    //create transport resources of user-endpoints later
    if (rqos.m_reliability.kind == RELIABLE_RELIABILITY_QOS) {
        (void)createSendResources(reader);
    }
    bool request_unique_flows = false;
    uint16_t initial_port = 0U;
    uint16_t final_port = 0U;
    if (!get_unique_flows_parameters(m_att, reader->getAttributes(), request_unique_flows, initial_port, final_port)) {
        return false;
    }
    if (!createAndAssociateReceiverswithEndpoint(reader, request_unique_flows, initial_port, final_port)) {
        logWarning(RTPS_PARTICIPANT, "create receiver endpoint failed when update local reader.");
    }
    return this->mp_builtinProtocols->updateLocalReader(reader, topicAtt, rqos, content_filter, notify);
}

/*
 *
 * AUXILIARY METHODS
 *
 *
 */

bool RTPSParticipantImpl::existsEntityId(const EntityId_t& ent, EndpointKind_t kind) const {
    auto check = [&ent](Endpoint* e) {
        return ent == e->getGuid().entityId;
    };

    vbs::shared_lock<vbs::shared_mutex> _(endpoints_list_mutex);

    if (kind == WRITER) {
        return std::any_of(m_userWriterList.begin(), m_userWriterList.end(), check);
    } else {
        return std::any_of(m_userReaderList.begin(), m_userReaderList.end(), check);
    }
}

/*
 *
 * RECEIVER RESOURCE METHODS
 *
 */
bool RTPSParticipantImpl::assignEndpointListenResources(Endpoint* endp) {
    // Tag the endpoint with the ReceiverResources
    bool valid = true;

    /* No need to check for emptiness on the lists, as it was already done on part function
       In case are using the default list of Locators they have already been embedded to the
       parameters
     */

    // UNICAST
    (void)assignEndpoint2LocatorList(endp, endp->getAttributes().unicastLocatorList);
    // MULTICAST
    (void)assignEndpoint2LocatorList(endp, endp->getAttributes().multicastLocatorList);
    return valid;
}

bool RTPSParticipantImpl::createAndAssociateReceiverswithEndpoint(Endpoint* pend, bool unique_flows,
                                                                  uint16_t initial_unique_port,
                                                                  uint16_t final_unique_port) {
    /*	This function...
        - Asks the network factory for new resources
        - Encapsulates the new resources within the ReceiverControlBlock list
        - Associated the endpoint to the new elements in the list
        - Launches the listener thread
     */

    auto& attributes = pend->getAttributes();
    if (unique_flows) {
        attributes.multicastLocatorList.clear();
        attributes.unicastLocatorList = m_att.defaultUnicastLocatorList;
        attributes.external_unicast_locators.clear();

        uint16_t port = initial_unique_port;
        while (port < final_unique_port) {
            // Set port on unicast locators
            for (Locator_t& loc : attributes.unicastLocatorList) {
                loc.port = port;
            }

            // Try creating receiver resources
            if (createReceiverResources(attributes.unicastLocatorList, false, true)) {
                break;
            }

            // Try with next port
            ++port;
        }

        // Fail when unique ports are exhausted
        if (port >= final_unique_port) {
            elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_OUT_OF_RESOURCES,  //LCOV_EXCL_START
                      "Unique flows requested but exhausted. Port range: " << initial_unique_port << "-"
                                                                           << final_unique_port);
            return false;
        }  //LCOV_EXCL_STOP
    } else {
        // 1 - Ask the network factory to generate the elements that do still not exist
        // Iterate through the list of unicast and multicast locators the endpoint has... unless its
        // empty In that case, just use the standard
        if (attributes.unicastLocatorList.empty() && attributes.multicastLocatorList.empty()) {
            // Take default locators from the participant.
            attributes.unicastLocatorList = m_att.defaultUnicastLocatorList;
            attributes.multicastLocatorList = m_att.defaultMulticastLocatorList;
            attributes.external_unicast_locators = m_att.default_external_unicast_locators;
        }
        (void)createReceiverResources(attributes.unicastLocatorList, false, true);
        (void)createReceiverResources(attributes.multicastLocatorList, false, true);
    }

    edds::rtps::ExternalLocatorsProcessor::set_listening_locators(attributes.external_unicast_locators,
                                                                  attributes.unicastLocatorList);

    // Associate the Endpoint with ReceiverControlBlock
    (void)assignEndpointListenResources(pend);
    return true;
}

bool RTPSParticipantImpl::assignEndpoint2LocatorList(Endpoint* endp, LocatorList_t& list) {
    /* Note:
       The previous version of this function associated (or created) ListenResources and added the
       endpoint to them. It then requested the list of Locators the Listener is listening to and
       appended to the LocatorList_t from the parameters.

       This has been removed because it is considered redundant. For ReceiveResources that listen on
       multiple interfaces, only one of the supported Locators is needed to make the match, and the
       case of new ListenResources being created has been removed since its the NetworkFactory the
       one that takes care of Resource creation.
     */
    for (auto lit = list.begin(); lit != list.end(); ++lit) {
        // Iteration of all Locators within the Locator list passed down as argument
        vbs::shared_lock<vbs::shared_mutex> guard(m_receiverResourcelistMutex);
        // Check among ReceiverResources whether the locator is supported or not
        for (auto it = m_receiverResourcelist.begin(); it != m_receiverResourcelist.end(); ++it) {
            // Take mutex for the resource since we are going to interact with shared resources
            // std::lock_guard<std::mutex> guard((*it).mtx);
            if (it->Receiver->SupportsLocator(*lit)) {
                // Supported! Take mutex and update lists - We maintain reader/writer discrimination
                // just in case
                logDebug(RTPS_PARTICIPANT, "receiver associate endpoint " << endp->getGuid());
                it->mp_receiver->associateEndpoint(endp);
                // end association between reader/writer and the receive resources
            }
        }
        // Finished iteratig through all ListenResources for a single Locator (from the parameter
        // list). Since this function is called after checking with NetFactory we do not have to
        // create any more resource.
    }
    return true;
}

bool RTPSParticipantImpl::createReceiverResources(LocatorList_t& Locator_list, bool ApplyMutation,
                                                  bool RegisterReceiver, int32_t kind) {
    bool ret_val = Locator_list.empty();
    std::shared_ptr<ReceiverResource> pReceiver;

    if (ret_val) {
        return ret_val;
    }
    if (!m_transportFactory) {
        return false;
    }
#if HAVE_SECURITY
    // An auxilary buffer is needed in the ReceiverResource to to decrypt the message,
    // that imposes a limit in the received messages size even if the transport allows (uint32_t)
    // messages size.
    uint32_t max_receiver_buffer_size =
        is_secure() ? std::numeric_limits<uint16_t>::max() : std::numeric_limits<uint32_t>::max();
#else
    uint32_t max_receiver_buffer_size = std::numeric_limits<uint32_t>::max();
#endif  // if HAVE_SECURITY

    for (auto it_loc = Locator_list.begin(); it_loc != Locator_list.end(); ++it_loc) {
        if (kind == LOCATOR_KIND_INVALID || (*it_loc).kind == kind) {
            uint32_t tries = 0U;
            bool already_open = false;
            pReceiver = m_transportFactory->BuildReceiverResources(get_transport_name(), *it_loc,
                                                                   max_receiver_buffer_size, already_open);
            if (!already_open && (!pReceiver) && ApplyMutation) {
                while (!already_open && (!pReceiver) && (tries < m_att.builtin.mutation_tries)) {
                    tries++;
                    *it_loc = applyLocatorAdaptRule(*it_loc);
                    pReceiver = m_transportFactory->BuildReceiverResources(get_transport_name(), *it_loc,
                                                                           max_receiver_buffer_size, already_open);
                }
            }
            if (pReceiver != nullptr) {
                ret_val = true;
                std::lock_guard<vbs::shared_mutex> lock(m_receiverResourcelistMutex);
                // Push the new items into the ReceiverResource buffer
                m_receiverResourcelist.emplace_back(pReceiver);
                // Create and init the MessageReceiver
                auto mr = new MessageReceiver(this, pReceiver->max_message_size());
                if (kind == LOCATOR_KIND_RAW) {
                    mr->set_compatible();
                }
                m_receiverResourcelist.back().mp_receiver = mr;
                // Start reception
                if (RegisterReceiver) {
                    m_receiverResourcelist.back().Receiver->RegisterReceiver(mr);
                }
                pReceiver = nullptr;

            } else if (already_open) {
                ret_val = true;
            } else if (!already_open) {
                logDebug(RTPS_PARTICIPANT, "ReceiverResource create fail, locator: " << *it_loc << " try:" << tries);
            }
        }
    }

    return ret_val;
}

bool RTPSParticipantImpl::createSendResources(Endpoint* pend) {
    if (pend->m_att.remoteLocatorList.empty()) {
        // Adds the default locators of every registered transport.
        m_transportFactory->GetDefaultOutputLocators(get_transport_name(), pend->m_att.remoteLocatorList);
    }
    createSenderResources(pend->m_att.remoteLocatorList);
    return true;
}

void RTPSParticipantImpl::createSenderResources(const LocatorList_t& locator_list) {
    for (auto it_loc = locator_list.begin(); it_loc != locator_list.end(); ++it_loc) {
        (void)m_transportFactory->BuildSendResources(get_transport_name(), *it_loc);
    }
}

void RTPSParticipantImpl::createSenderResources(const Locator_t& locator) {
    (void)m_transportFactory->BuildSendResources(get_transport_name(), locator);
}

bool RTPSParticipantImpl::deleteUserEndpoint(const GUID_t& endpoint) {
    if (getGuid().guidPrefix != endpoint.guidPrefix) {
        return false;
    }

    bool found = false, found_in_users = false;
    Endpoint* p_endpoint = nullptr;

    if (endpoint.entityId.is_writer()) {
        std::lock_guard<vbs::shared_mutex> _(endpoints_list_mutex);

        for (auto wit = m_userWriterList.begin(); wit != m_userWriterList.end(); ++wit) {
            if ((*wit)->getGuid().entityId == endpoint.entityId)  // Found it
            {
                (void)m_userWriterList.erase(wit);
                found_in_users = true;
                break;
            }
        }

        for (auto wit = m_allWriterList.begin(); wit != m_allWriterList.end(); ++wit) {
            if ((*wit)->getGuid().entityId == endpoint.entityId)  // Found it
            {
                p_endpoint = *wit;
                (void)m_allWriterList.erase(wit);
                found = true;
                break;
            }
        }
    } else {
        std::lock_guard<vbs::shared_mutex> _(endpoints_list_mutex);

        for (auto rit = m_userReaderList.begin(); rit != m_userReaderList.end(); ++rit) {
            if ((*rit)->getGuid().entityId == endpoint.entityId)  // Found it
            {
                (void)m_userReaderList.erase(rit);
                found_in_users = true;
                break;
            }
        }

        for (auto rit = m_allReaderList.begin(); rit != m_allReaderList.end(); ++rit) {
            if ((*rit)->getGuid().entityId == endpoint.entityId)  // Found it
            {
                p_endpoint = *rit;
                (void)m_allReaderList.erase(rit);
                found = true;
                break;
            }
        }
    }

    if (!found) {
        return false;
    }

    {
        vbs::shared_lock<vbs::shared_mutex> _(m_receiverResourcelistMutex);

        for (auto& rb : m_receiverResourcelist) {
            auto receiver = rb.mp_receiver;
            if (receiver) {
                receiver->removeEndpoint(p_endpoint);
            }
        }
    }

    // REMOVE FROM BUILTIN PROTOCOLS
    if (p_endpoint->getAttributes().endpointKind == WRITER) {
        if (found_in_users) {
            (void)mp_builtinProtocols->removeLocalWriter(static_cast<RTPSWriter*>(p_endpoint));
        }

#if HAVE_SECURITY
        if (p_endpoint->getAttributes().security_attributes().is_submessage_protected ||
            p_endpoint->getAttributes().security_attributes().is_payload_protected) {
            (void)m_security_manager.unregister_local_writer(p_endpoint->getGuid());
        }
#endif  // if HAVE_SECURITY
    } else {
        if (found_in_users) {
            (void)mp_builtinProtocols->removeLocalReader(static_cast<RTPSReader*>(p_endpoint));
        }

#if HAVE_SECURITY
        if (p_endpoint->getAttributes().security_attributes().is_submessage_protected ||
            p_endpoint->getAttributes().security_attributes().is_payload_protected) {
            (void)m_security_manager.unregister_local_reader(p_endpoint->getGuid());
        }
#endif  // if HAVE_SECURITY
    }

    delete (p_endpoint);
    return true;
}

void RTPSParticipantImpl::deleteAllUserEndpoints() {
    std::vector<Endpoint*> tmp(0U);

    {
        using namespace std;

        lock_guard<vbs::shared_mutex> _(endpoints_list_mutex);

        // move the collections to a local
        tmp.resize(m_userWriterList.size() + m_userReaderList.size());
        auto it = move(m_userWriterList.begin(), m_userWriterList.end(), tmp.begin());
        it = move(m_userReaderList.begin(), m_userReaderList.end(), it);

        // check we have copied all elements
        assert(tmp.end() == it);

        // now update the all collections by removing the user elements
        sort(m_userWriterList.begin(), m_userWriterList.end());
        sort(m_userReaderList.begin(), m_userReaderList.end());
        sort(m_allWriterList.begin(), m_allWriterList.end());
        sort(m_allReaderList.begin(), m_allReaderList.end());

        vector<RTPSWriter*> writers;
        (void)set_difference(m_allWriterList.begin(), m_allWriterList.end(), m_userWriterList.begin(),
                             m_userWriterList.end(), back_inserter(writers));
        swap(writers, m_allWriterList);

        vector<RTPSReader*> readers;
        (void)set_difference(m_allReaderList.begin(), m_allReaderList.end(), m_userReaderList.begin(),
                             m_userReaderList.end(), back_inserter(readers));
        swap(readers, m_allReaderList);

        // remove dangling references
        m_userWriterList.clear();
        m_userReaderList.clear();
    }

    // unlink the transport receiver blocks from the endpoints
    for (auto endpoint : tmp) {
        vbs::shared_lock<vbs::shared_mutex> _(m_receiverResourcelistMutex);

        for (auto& rb : m_receiverResourcelist) {
            auto receiver = rb.mp_receiver;
            if (receiver) {
                receiver->removeEndpoint(endpoint);
            }
        }
    }

    // Remove from builtin protocols
    auto removeEndpoint = [this](EndpointKind_t kind, Endpoint* p) {
        return kind == WRITER ? mp_builtinProtocols->removeLocalWriter((RTPSWriter*)p)
                              : mp_builtinProtocols->removeLocalReader((RTPSReader*)p);
    };

#if HAVE_SECURITY
    bool (evbs::ertps::rtps::security::SecurityManager::*unregister_endpoint[2])(const GUID_t& writer_guid);
    unregister_endpoint[WRITER] = &security::SecurityManager::unregister_local_writer;
    unregister_endpoint[READER] = &security::SecurityManager::unregister_local_reader;
#endif  // if HAVE_SECURITY

    for (auto endpoint : tmp) {
        auto kind = endpoint->getAttributes().endpointKind;
        removeEndpoint(kind, endpoint);

#if HAVE_SECURITY
        if (endpoint->getAttributes().security_attributes().is_submessage_protected ||
            endpoint->getAttributes().security_attributes().is_payload_protected) {
            (void)(m_security_manager.*unregister_endpoint[kind])(endpoint->getGuid());
        }
#endif  // if HAVE_SECURITY

        // remove the endpoints
        delete (endpoint);
    }
}

void RTPSParticipantImpl::normalize_endpoint_locators(EndpointAttributes& endpoint_att) {
    // Locators with port 0, calculate port.
    uint32_t unicast_port = calculate_well_known_port(domain_id_, m_att, false);
    for (Locator_t& loc : endpoint_att.unicastLocatorList) {
        (void)m_transportFactory->fill_default_locator_port(get_transport_name(), loc, unicast_port);
    }
    uint32_t multicast_port = calculate_well_known_port(domain_id_, m_att, true);
    for (Locator_t& loc : endpoint_att.multicastLocatorList) {
        (void)m_transportFactory->fill_default_locator_port(get_transport_name(), loc, multicast_port);
    }

    // Normalize unicast locators
    if (!endpoint_att.unicastLocatorList.empty()) {
        m_transportFactory->NormalizeLocators(get_transport_name(), endpoint_att.unicastLocatorList);
    }
}

std::vector<std::string> RTPSParticipantImpl::getParticipantNames() const {
    std::vector<std::string> participant_names;
    auto pdp = mp_builtinProtocols->mp_PDP;
    for (auto it = pdp->ParticipantProxiesBegin(); it != pdp->ParticipantProxiesEnd(); ++it) {
        participant_names.emplace_back((*it)->m_participantName.to_string());
    }
    return participant_names;
}

void RTPSParticipantImpl::setGuid(GUID_t& guid) {
    m_guid = guid;
}

void RTPSParticipantImpl::announceRTPSParticipantState() {
    return mp_builtinProtocols->announceRTPSParticipantState();
}

void RTPSParticipantImpl::stopRTPSParticipantAnnouncement() {
    return mp_builtinProtocols->stopRTPSParticipantAnnouncement();
}

void RTPSParticipantImpl::resetRTPSParticipantAnnouncement() {
    return mp_builtinProtocols->resetRTPSParticipantAnnouncement();
}

bool RTPSParticipantImpl::startRTPSParticipantInitialAnnouncement(uint32_t count, Duration_t period) {
    return mp_builtinProtocols->startRTPSParticipantInitialAnnouncement(count, period);
}

EvbsVersion_t RTPSParticipantImpl::get_remote_evbsversion() {
    return m_evbsRemoteVersion;
}

void RTPSParticipantImpl::loose_next_change() {
    // NOTE: This is replaced by the test transport
    // this->mp_send_thr->loose_next_change();
}

bool RTPSParticipantImpl::newRemoteEndpointDiscovered(const GUID_t& pguid, int16_t userDefinedId, EndpointKind_t kind) {
    if ((m_att.builtin.discovery_config.discoveryProtocol != DiscoveryProtocol::SIMPLE) ||
        (m_att.builtin.discovery_config.use_STATIC_EndpointDiscoveryProtocol == false)) {
        logWarning(RTPS_PARTICIPANT,
                   "Remote Endpoints can only be activated with static discovery "
                   "protocol over PDP simple protocol");
        return false;
    }

    if (PDPSimple* pS = dynamic_cast<PDPSimple*>(mp_builtinProtocols->mp_PDP)) {
        return pS->newRemoteEndpointStaticallyDiscovered(pguid, userDefinedId, kind);
    }

    return false;
}

// void RTPSParticipantImpl::ResourceSemaphorePost() {
//     if (mp_ResourceSemaphore != nullptr) {
//         mp_ResourceSemaphore->post();
//     }
// }

// void RTPSParticipantImpl::ResourceSemaphoreWait() {
//     if (mp_ResourceSemaphore != nullptr) {
//         mp_ResourceSemaphore->wait();
//     }
// }

void RTPSParticipantImpl::assert_remote_participant_liveliness(const GuidPrefix_t& remote_guid) {
    if (mp_builtinProtocols && mp_builtinProtocols->mp_PDP) {
        mp_builtinProtocols->mp_PDP->assert_remote_participant_liveliness(remote_guid);
    }
}

/**
 * Get the list of locators from which this publisher may send data.
 *
 * @param [out] locators  LocatorList_t where the list of locators will be stored.
 */
void RTPSParticipantImpl::get_sending_locators(rtps::LocatorList_t& locators) const {
    locators.clear();
    SendResourceList& send_resource_list =
        m_transportFactory->getSendResource(get_transport_name(), vbs::transport::TransmitorType::TRANSMITOR_TYPE_UDP);

    // Traverse the sender list and query
    for (const auto& send_resource : send_resource_list) {
        send_resource->add_locators_to_list(locators);
    }
}
/**
 * Get the list of locators from which this publisher may send data.
 *
 * @param [out] locators  LocatorList_t where the list of locators will be stored.
 */
const Locator_t& RTPSParticipantImpl::find_optimal_remote_locator(
    const ResourceLimitedVector<Locator_t>& locators) const {
    rtps::Locator_t locator_any;
    rtps::LocatorList_t locs;

    locator_any.kind = LOCATOR_KIND_UDPv4;
    (void)IPLocator::setIPv4(locator_any, "0.0.0.0");

    SendResourceList& send_resource_list =
        m_transportFactory->getSendResource(get_transport_name(), vbs::transport::TransmitorType::TRANSMITOR_TYPE_UDP);

    // Traverse the sender list and query
    for (auto& send_resource : send_resource_list) {
        if (!send_resource->check_ip_address(locator_any)) {
            send_resource->add_locators_to_list(locs);
            auto loc = *locs.begin();

            for (auto& locator : locators) {
                if (IPLocator::isSameNetworkSegment(IPLocator::toIPv4string(loc), IPLocator::toIPv4string(locator),
                                                    loc.mask_len)) {
                    return locator;
                }
            }
        }
        locs.clear();
    }
    return *locators.begin();
}
/*remote locator is not same netsegment with any send resource, ignore....*/
bool RTPSParticipantImpl::is_valid_remote_locator(rtps::Locator_t& locator) const {
    if (locator.kind != LOCATOR_KIND_UDPv4)
        return true;

    if (network_factory()->is_local_locator(get_transport_name(), locator)) {
        return true;
    }

    rtps::LocatorList_t sendlocatorlist;
    // Traverse the sender list and query
    get_sending_locators(sendlocatorlist);

    for (auto& loc : sendlocatorlist) {
        if (loc.mask_len > 0) {
            if (IPLocator::isSameNetworkSegment(IPLocator::toIPv4string(loc), IPLocator::toIPv4string(locator),
                                                loc.mask_len)) {
                return true;
            }
        }
    }
    logDebug(RTPS_PDP, "Remote locator " << locator << " FILTERED");

    return false;
}

uint32_t RTPSParticipantImpl::getMaxMessageSize() const {
#if HAVE_SECURITY
    // An auxilary buffer is needed in the ReceiverResource to to decrypt the message,
    // that imposes a limit in the received messages size even if the transport allows (uint32_t)
    // messages size. So the sender limits also its size.
    uint32_t max_receiver_buffer_size =
        is_secure() ? std::numeric_limits<uint16_t>::max() : std::numeric_limits<uint32_t>::max();
#else
    uint32_t max_receiver_buffer_size = std::numeric_limits<uint32_t>::max();
#endif  // if HAVE_SECURITY

    return (std::min)({max_output_message_size_, m_maxMessageSize, max_receiver_buffer_size});
}

uint32_t RTPSParticipantImpl::getMaxDataSize() {
    return calculateMaxDataSize(getMaxMessageSize());
}

uint32_t RTPSParticipantImpl::calculateMaxDataSize(uint32_t length) {
    uint32_t maxDataSize = length;

#if HAVE_SECURITY
    // If there is rtps messsage protection, reduce max size for messages,
    // because extra data is added on encryption.
    if (security_attributes_.is_rtps_protected) {
        maxDataSize -= m_security_manager.calculate_extra_size_for_rtps_message();
    }
#endif  // if HAVE_SECURITY

    // RTPS header
    maxDataSize -= RTPSMESSAGE_HEADER_SIZE;
    return maxDataSize;
}

#if HAVE_SECURITY
bool RTPSParticipantImpl::pairing_remote_reader_with_local_writer_after_security(
    const GUID_t& local_writer, const ReaderProxyData& remote_reader_data) {
    bool return_value;

    return_value = mp_builtinProtocols->mp_PDP->pairing_remote_reader_with_local_writer_after_security(
        local_writer, remote_reader_data);
    if ((!return_value) && (mp_builtinProtocols->mp_WLP != nullptr)) {
        return_value = mp_builtinProtocols->mp_WLP->pairing_remote_reader_with_local_writer_after_security(
            local_writer, remote_reader_data);
    }

    return return_value;
}

bool RTPSParticipantImpl::pairing_remote_writer_with_local_reader_after_security(
    const GUID_t& local_reader, const WriterProxyData& remote_writer_data) {
    bool return_value;

    return_value = mp_builtinProtocols->mp_PDP->pairing_remote_writer_with_local_reader_after_security(
        local_reader, remote_writer_data);
    if ((!return_value) && (mp_builtinProtocols->mp_WLP != nullptr)) {
        return_value = mp_builtinProtocols->mp_WLP->pairing_remote_writer_with_local_reader_after_security(
            local_reader, remote_writer_data);
    }

    return return_value;
}

bool RTPSParticipantImpl::is_security_enabled_for_writer(const WriterAttributes& writer_attributes) {
    if ((!is_initialized()) || (!is_secure())) {
        return false;
    }

    if (security_attributes().is_rtps_protected) {
        return true;
    }

    security::EndpointSecurityAttributes security_attributes;
    if (security_manager().get_datawriter_sec_attributes(writer_attributes.endpoint.properties, security_attributes)) {
        return ((security_attributes.is_payload_protected == true) ||
                (security_attributes.is_submessage_protected == true));
    }

    return false;
}

bool RTPSParticipantImpl::is_security_enabled_for_reader(const ReaderAttributes& reader_attributes) {
    if ((!is_initialized()) || (!is_secure())) {
        return false;
    }

    if (security_attributes().is_rtps_protected) {
        return true;
    }

    security::EndpointSecurityAttributes security_attributes;
    if (security_manager().get_datareader_sec_attributes(reader_attributes.endpoint.properties, security_attributes)) {
        return ((security_attributes.is_payload_protected == true) ||
                (security_attributes.is_submessage_protected == true));
    }

    return false;
}

#endif  // if HAVE_SECURITY

PDP* RTPSParticipantImpl::pdp() {
    return mp_builtinProtocols->mp_PDP;
}

PDPSimple* RTPSParticipantImpl::pdpsimple() {
    return dynamic_cast<PDPSimple*>(mp_builtinProtocols->mp_PDP);
}

WLP* RTPSParticipantImpl::wlp() {
    return mp_builtinProtocols->mp_WLP;
}

edds::dds::builtin::LookupManager* RTPSParticipantImpl::lookup_manager() const {
    return mp_builtinProtocols->tlm_;
}

IPersistenceService* RTPSParticipantImpl::get_persistence_service(const EndpointAttributes& param) {
    IPersistenceService* ret_val;

    ret_val = PersistenceFactory::create_persistence_service(param.properties);
    return ret_val != nullptr ? ret_val : PersistenceFactory::create_persistence_service(m_att.properties);
}

bool RTPSParticipantImpl::get_persistence_service(bool is_builtin, const EndpointAttributes& param,
                                                  IPersistenceService*& service) {
    service = nullptr;

    const char* debug_label = (param.endpointKind == WRITER ? "writer" : "reader");

    // Check if also support persistence with TRANSIENT_LOCAL.
    DurabilityKind_t durability_red_line = get_persistence_durability_red_line(is_builtin);
    if (param.durabilityKind >= durability_red_line) {
        if (param.persistence_guid == c_Guid_Unknown) {
            elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_BAD_PARAMETER,
                      "Cannot create persistence service. Persistence GUID not specified");
            return false;
        }
        service = get_persistence_service(param);
        if (service == nullptr) {
            elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_ERROR,
                      "Couldn't create writer persistence service for transient/persistent " << debug_label);
            return false;
        }
    }

    // Error log level can be disable. Avoid unused warning
    static_cast<void>(debug_label);

    return true;
}

bool RTPSParticipantImpl::get_new_entity_id(EntityId_t& entityId) {
    if (entityId == c_EntityId_Unknown) {
        uint32_t idnum = ++IdCounter;
        octet* c = reinterpret_cast<octet*>(&idnum);
        entityId.value[2] = c[0];
        entityId.value[1] = c[1];
        entityId.value[0] = c[2];
        entityId.value[3] = 0x01U;  // Vendor specific
    } else {
        return ((!existsEntityId(entityId, READER)) && (!existsEntityId(entityId, WRITER)));
    }

    return true;
}

void RTPSParticipantImpl::set_check_type_function(std::function<bool(const std::string&)>&& check_type) {
    type_check_fn_ = std::move(check_type);
}

//! Compare metatraffic locators list searching for mutations
bool RTPSParticipantImpl::did_mutation_took_place_on_meta(const LocatorList_t& MulticastLocatorList,
                                                          const LocatorList_t& UnicastLocatorList) const {
    using namespace std;
    using namespace evbs::edds::rtps;

    if (m_att.builtin.metatrafficMulticastLocatorList == MulticastLocatorList &&
        m_att.builtin.metatrafficUnicastLocatorList == UnicastLocatorList) {
        // no mutation
        return false;
    }

    // If one of the locators is 0.0.0.0 we must replace it by all local interfaces like the
    // framework does
    list<Locator_t> unicast_real_locators;
    LocatorListConstIterator it = UnicastLocatorList.begin(), old_it;
    LocatorList_t locals;

    do {
        // copy ordinary locators till the first ANY
        old_it = it;
        it = find_if(it, UnicastLocatorList.end(), IPLocator::isAny);

        // copy ordinary locators
        (void)copy(old_it, it, back_inserter(unicast_real_locators));

        // transform new ones if needed
        if (it != UnicastLocatorList.end()) {
            const Locator_t& an_any = *it;

            // load interfaces if needed
            if (locals.empty()) {
                (void)IPFinder::getIP4Address(&locals);
            }

            // add a locator for each local
            (void)transform(locals.begin(), locals.end(), back_inserter(unicast_real_locators),
                            [&an_any](const Locator_t& loc) -> Locator_t {
                                Locator_t specific(loc);
                                specific.port = an_any.port;
                                specific.kind = an_any.kind;
                                return specific;
                            });

            // search for the next if any
            ++it;
        }
    } while (it != UnicastLocatorList.end());

#ifdef EDDS_TCP_TRANSPORT_COMPILE
    // TCP is a special case because physical ports are taken from the TransportDescriptors
    // besides WAN address may be added by the transport
    struct ResetLogical {
        // use of unary_function to introduce the following aliases is deprecated
        // using argument_type = Locator_t;
        // using result_type   = Locator_t&;

        using Transports = vector<shared_ptr<TransportDescriptorInterface>>;

        ResetLogical(const Transports& tp) : Transports_(tp) {
            for (auto& desc : Transports_) {
                if (nullptr == tcp4) {
                    tcp4 = dynamic_pointer_cast<TCPv4TransportDescriptor>(desc);
                }
#ifdef VBS_IPV6
                if (nullptr == tcp6) {
                    tcp6 = dynamic_pointer_cast<TCPv6TransportDescriptor>(desc);
                }
#endif
            }
        }

        uint16_t Tcp4ListeningPort() const {
            return tcp4 ? (tcp4->listening_ports.empty() ? 0 : tcp4->listening_ports[0]) : 0;
        }
#ifdef VBS_IPV6
        uint16_t Tcp6ListeningPort() const {
            return tcp6 ? (tcp6->listening_ports.empty() ? 0 : tcp6->listening_ports[0]) : 0;
        }
#endif
        void set_wan_address(Locator_t& loc) const {
            if (tcp4) {
                assert(LOCATOR_KIND_TCPv4 == loc.kind);
                auto& ip = tcp4->wan_addr;
                (void)IPLocator::setWan(loc, ip[0], ip[1], ip[2], ip[3]);
            }
        }

        Locator_t operator()(const Locator_t& loc) const {
            Locator_t ret(loc);
            switch (loc.kind) {
                case LOCATOR_KIND_TCPv4:
                    set_wan_address(ret);
                    (void)IPLocator::setPhysicalPort(ret, Tcp4ListeningPort());
                    break;
#ifdef VBS_IPV6
                case LOCATOR_KIND_TCPv6:
                    (void)IPLocator::setPhysicalPort(ret, Tcp6ListeningPort());
                    break;
#endif
            }
            return ret;
        }

        // reference to the transports
        const Transports& Transports_;
        shared_ptr<TCPv4TransportDescriptor> tcp4;
#ifdef VBS_IPV6
        shared_ptr<TCPv6TransportDescriptor> tcp6;
#endif
    } transform_functor(m_att.userTransports);

    // transform-copy
    set<Locator_t> update_attributes;

    (void)transform(m_att.builtin.metatrafficMulticastLocatorList.begin(),
                    m_att.builtin.metatrafficMulticastLocatorList.end(),
                    inserter(update_attributes, update_attributes.begin()), transform_functor);

    (void)transform(m_att.builtin.metatrafficUnicastLocatorList.begin(),
                    m_att.builtin.metatrafficUnicastLocatorList.end(),
                    inserter(update_attributes, update_attributes.begin()), transform_functor);

    set<Locator_t> original_ones;

    (void)transform(MulticastLocatorList.begin(), MulticastLocatorList.end(),
                    inserter(original_ones, original_ones.begin()), transform_functor);

    (void)transform(unicast_real_locators.begin(), unicast_real_locators.end(),
                    inserter(original_ones, original_ones.begin()), transform_functor);

    // if equal then no mutation took place on physical ports
    return !(update_attributes == original_ones);
#else
    return false;
#endif
}

DurabilityKind_t RTPSParticipantImpl::get_persistence_durability_red_line(bool is_builtin_endpoint) {
    DurabilityKind_t durability_red_line = TRANSIENT;
    if (!is_builtin_endpoint) {
        std::string* persistence_support_transient_local_property =
            PropertyPolicyHelper::find_property(m_att.properties, "dds.persistence.also-support-transient-local");
        if ((nullptr != persistence_support_transient_local_property) &&
            (0 == persistence_support_transient_local_property->compare("true"))) {
            durability_red_line = TRANSIENT_LOCAL;
        }
    }

    return durability_red_line;
}

void RTPSParticipantImpl::environment_file_has_changed() {
    RTPSParticipantAttributes patt = m_att;

    {
        logWarning(RTPS_QOS_CHECK, "Trying to add Discovery Servers to a participant which is not a SERVER, BACKUP "
                                       << "or an overriden CLIENT (SIMPLE participant transformed into CLIENT with the "
                                          "environment variable)");
    }
}

/**
 * Invalidate all selector entries containing certain multicast locator.
 *
 * This function will process all entries from 'index' onwards and, if any
 * of them has 'locator' on its multicast list, will invalidate them
 * (i.e. their 'transport_should_process' flag will be changed to false).
 *
 * If this function returns true, the locator received should be selected.
 *
 * @param entries   Selector entries collection to process
 * @param index     Starting index to process
 * @param locator   Locator to be searched
 *
 * @return true when at least one entry was invalidated, false otherwise
 */
static bool check_and_invalidate(ertps::ResourceLimitedVector<LocatorSelectorEntry*>& entries, size_t index,
                                 const Locator& locator) {
    bool ret_val = false;
    for (; index < entries.size(); ++index) {
        LocatorSelectorEntry* entry = entries[index];
        if (entry->transport_should_process) {
            for (const Locator& loc : entry->multicast) {
                if (loc == locator) {
                    entry->transport_should_process = false;
                    ret_val = true;
                    break;
                }
            }
        }
    }

    return ret_val;
}

void RTPSParticipantImpl::select_locators(LocatorSelector& selector) {
    selector.selection_start();

    ertps::ResourceLimitedVector<LocatorSelectorEntry*>& entries = selector.transport_starts();

    for (size_t i = 0; i < entries.size(); ++i) {
        LocatorSelectorEntry* entry = entries[i];
        if (entry->transport_should_process) {
            bool selected = false;

            // First try to find a multicast locator which is at least on another list.
            for (size_t j = 0; j < entry->multicast.size() && (!selected); ++j) {
                if (m_transportFactory->select_multicast(get_transport_name(), entry->multicast[j])) {
                    if (check_and_invalidate(entries, i + 1, entry->multicast[j])) {
                        (void)entry->state.multicast.push_back(j);
                        selected = true;
                    } else if (entry->unicast.size() == 0) {
                        (void)entry->state.multicast.push_back(j);
                        selected = true;
                    }
                }
            }

            // If we couldn't find a multicast locator, select all unicast locators
            if (!selected) {
                for (size_t j = 0; j < entry->unicast.size(); ++j) {
                    if (!selector.is_selected(entry->unicast[j])) {
                        (void)entry->state.unicast.push_back(j);
                        selected = true;
                    }
                }
            }

            // Select this entry if necessary
            if (selected) {
                selector.select(i);
            }
        }
    }
}

EvbsVersion_t RTPSParticipantImpl::get_version_by_guid(const GuidPrefix_t& guid_prefix) {
    if (mp_builtinProtocols) {
        return mp_builtinProtocols->get_version_by_guid(guid_prefix);
    }
    return evbs::edds::rtps::c_Version_vbs_unknown;
}

bool RTPSParticipantImpl::is_participant_ignored(const GuidPrefix_t& participant_guid) {
    vbs::shared_lock<vbs::shared_mutex> _(ignored_mtx_);
    return ignored_participants_.find(participant_guid) != ignored_participants_.end();
}

bool RTPSParticipantImpl::is_writer_ignored(const GUID_t& /*writer_guid*/) {  //LCOV_EXCL_START
    return false;
}

bool RTPSParticipantImpl::is_reader_ignored(const GUID_t& /*reader_guid*/) {
    return false;
}  //LCOV_EXCL_STOP

bool RTPSParticipantImpl::ignore_participant(const GuidPrefix_t& participant_guid) {
    if (participant_guid == m_guid.guidPrefix) {
        elogWarning(RTPS_PARTICIPANT, "A participant is unable to ignore itself");
        return false;
    }
    {
        std::unique_lock<vbs::shared_mutex> _(ignored_mtx_);
        ignored_participants_.insert(participant_guid);
    }
    pdp()->remove_remote_participant(GUID_t(participant_guid, c_EntityId_RTPSParticipant),
                                     ParticipantDiscoveryInfo::IGNORED_PARTICIPANT);

    return true;
}

bool RTPSParticipantImpl::ignore_writer(const GUID_t& /*writer_guid*/) {  //LCOV_EXCL_START
    return false;
}

bool RTPSParticipantImpl::ignore_reader(const GUID_t& /*reader_guid*/) {
    return false;
}  //LCOV_EXCL_STOP

std::string RTPSParticipantImpl::get_transport_name() const {
    std::stringstream ss;
    ss << "rtps_";
    for (size_t i = 0; i < m_guid.guidPrefix.size; ++i) {
        if (i > 0) {
            ss << ".";
        }
        ss << std::hex << std::setw(2) << std::setfill('0') << static_cast<int>(m_guid.guidPrefix.value[i]);
    }
    return ss.str();
}

ReturnCode_t RTPSParticipantImpl::assert_liveliness(RTPSWriter* writer) {
    if (!wlp()->assert_liveliness(writer->getGuid(), writer->get_liveliness_kind(),
                                  writer->get_liveliness_lease_duration())) {
        elogError(DATA_WRITER, RetCode_t::RETCODE_ERROR,  //LCOV_EXCL_START
                  "Could not assert liveliness of writer " << writer->getGuid());
        return ReturnCode_t::RETCODE_ERROR;
    }  //LCOV_EXCL_STOP

    if (writer->get_liveliness_kind() == MANUAL_BY_TOPIC_LIVELINESS_QOS) {
        // As described in the RTPS specification, if liveliness kind is manual a heartbeat must be
        // sent This only applies to stateful writers, as stateless writers do not send heartbeats

        StatefulWriter* stateful_writer = dynamic_cast<StatefulWriter*>(writer);

        if (stateful_writer != nullptr) {
            (void)stateful_writer->send_periodic_heartbeat(true, true);
        }
    }
    return ReturnCode_t::RETCODE_OK;
}

ReturnCode_t RTPSParticipantImpl::assert_liveliness() {
    if (wlp() != nullptr) {
        if (wlp()->assert_liveliness_manual_by_participant()) {
            return ReturnCode_t::RETCODE_OK;
        }
    } else {
        elogError(PARTICIPANT, RetCode_t::RETCODE_ERROR, "Invalid WLP, cannot assert liveliness of participant");
    }
    return ReturnCode_t::RETCODE_ERROR;
}

std::string RTPSParticipantImpl::get_db_info(bool log_record) {
    bool ret = false;
    uint32_t writer_num = 0;
    uint32_t reader_num = 0;
    std::vector<vbsutil::xmlparser::GuidPrefix_t> prefix_list;
    ret = get_participants_in_db(prefix_list);
    if (!ret) {
        elogInfo(PARTICIPANT, "prefix_list get failed! ");
        return "";
    }
    std::stringstream output;

    for (auto prefix : prefix_list) {
        evbs::ertps::rtps::ddb::DiscoveryParticipantInfo part_info;
        // Get the participant info by guid_prefix
        ret = get_participant_info(prefix, part_info);
        if (ret) {
            std::stringstream tmp_str;
            tmp_str << "prefix: " << prefix << " part_info: " << part_info;
            if (log_record) {
                elogInfo(PARTICIPANT, tmp_str.str());
            }
            output << tmp_str.str() << std::endl;
            for (auto reader : part_info.readers()) {
                evbs::ertps::rtps::ddb::DiscoveryEndpointInfo reader_info;
                ret = get_reader_info(reader, reader_info);
                if (ret) {
                    if (!reader_info.is_virtual()) {
                        reader_num++;
                    }
                    std::stringstream tmp_str;
                    tmp_str << "reader_info: " << reader_info;
                    if (log_record) {
                        elogInfo(PARTICIPANT, tmp_str.str());
                    }
                    output << tmp_str.str() << std::endl;
                }
            }
            for (auto writer : part_info.writers()) {
                evbs::ertps::rtps::ddb::DiscoveryEndpointInfo writer_info;
                ret = get_writer_info(writer, writer_info);
                if (ret) {
                    if (!writer_info.is_virtual()) {
                        writer_num++;
                    }
                    std::stringstream tmp_str;
                    tmp_str << "writer_info: " << writer_info;
                    if (log_record) {
                        elogInfo(PARTICIPANT, tmp_str.str());
                    }
                    output << tmp_str.str() << std::endl;
                }
            }
        }
    }
    std::stringstream tmp_str;
    tmp_str << "participant num: " << prefix_list.size() << " writer num: " << writer_num
            << " reader num: " << reader_num;
    elogInfo(PARTICIPANT, tmp_str.str());
    output << tmp_str.str() << std::endl;
    return output.str();
}

// Get all the participants in DataBase
bool RTPSParticipantImpl::get_participants_in_db(std::vector<GuidPrefix_t>& prefix_list) {
    return pdp()->get_participants_in_db(prefix_list);
}

// Get the participant info by guid_prefix
bool RTPSParticipantImpl::get_participant_info(const GuidPrefix_t& prefix, DiscoveryParticipantInfo& part_info) {
    return pdp()->get_participant_info(prefix, part_info);
}

// Get the reader info by guid
bool RTPSParticipantImpl::get_reader_info(const GUID_t& reader_guid, DiscoveryEndpointInfo& reader_info) {
    return pdp()->get_reader_info(reader_guid, reader_info);
}

// Get the writer info by guid
bool RTPSParticipantImpl::get_writer_info(const GUID_t& writer_guid, DiscoveryEndpointInfo& writer_info) {
    return pdp()->get_writer_info(writer_guid, writer_info);
}

bool RTPSParticipantImpl::get_topic_names(std::vector<std::string>& topic_list) {
    return pdp()->get_topic_names(topic_list);
}

bool RTPSParticipantImpl::has_writer(const std::string& topic_name) {
    return pdp()->has_writer(topic_name);
}

bool RTPSParticipantImpl::has_reader(const std::string& topic_name) {
    return pdp()->has_reader(topic_name);
}

bool RTPSParticipantImpl::get_writer_of_topic(const std::string& topic_name,
                                              std::vector<DiscoveryEndpointInfo>& writer_info_list) {
    return pdp()->get_writer_of_topic(topic_name, writer_info_list);
}

bool RTPSParticipantImpl::get_reader_of_topic(const std::string& topic_name,
                                              std::vector<DiscoveryEndpointInfo>& reader_info_list) {
    return pdp()->get_reader_of_topic(topic_name, reader_info_list);
}

bool RTPSParticipantImpl::has_participant(const std::string& participant_name) {
    return pdp()->has_participant(participant_name);
}

bool RTPSParticipantImpl::get_participant_of_name(const std::string& participant_name,
                                                  DiscoveryParticipantInfo& part_info) {
    return pdp()->get_participant_of_name(participant_name, part_info);
}

void RTPSParticipantImpl::registed_user_transport(vbs::RemoteEndpointInfo::LocatorType type, bool registed) {
    registed_user_transport_[type] = registed;
}

bool RTPSParticipantImpl::registed_user_transport(vbs::RemoteEndpointInfo::LocatorType type) {
    return registed_user_transport_[type];
}

} /* namespace rtps */
} /* namespace ertps */
} /* namespace evbs */
